Addressing review comments.

* list_dirs and get_dir_model added to base NBM class.
* Calling of list_dirs moved to handler.
* type=notebook added to notebook model.

IPython/html/services/notebooks/filenbmanager.py

@@ -1,459 +1,457 b''
 """A notebook manager that uses the local file system for storage.
 
 Authors:
 
 * Brian Granger
 * Zach Sailer
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import io
 import itertools
 import os
 import glob
 import shutil
 
 from tornado import web
 
 from .nbmanager import NotebookManager
 from IPython.nbformat import current
 from IPython.utils.traitlets import Unicode, Dict, Bool, TraitError
 from IPython.utils import tz
 
 #-----------------------------------------------------------------------------
 # Classes
 #-----------------------------------------------------------------------------
 
 class FileNotebookManager(NotebookManager):
     
     save_script = Bool(False, config=True,
         help="""Automatically create a Python script when saving the notebook.
         
         For easier use of import, %run and %load across notebooks, a
         <notebook-name>.py script will be created next to any
         <notebook-name>.ipynb on each save.  This can also be set with the
         short `--script` flag.
         """
     )
     
     checkpoint_dir = Unicode(config=True,
         help="""The location in which to keep notebook checkpoints
         
         By default, it is notebook-dir/.ipynb_checkpoints
         """
     )
     def _checkpoint_dir_default(self):
         return os.path.join(self.notebook_dir, '.ipynb_checkpoints')
     
     def _checkpoint_dir_changed(self, name, old, new):
         """do a bit of validation of the checkpoint dir"""
         if not os.path.isabs(new):
             # If we receive a non-absolute path, make it absolute.
             abs_new = os.path.abspath(new)
             self.checkpoint_dir = abs_new
             return
         if os.path.exists(new) and not os.path.isdir(new):
             raise TraitError("checkpoint dir %r is not a directory" % new)
         if not os.path.exists(new):
             self.log.info("Creating checkpoint dir %s", new)
             try:
                 os.mkdir(new)
             except:
                 raise TraitError("Couldn't create checkpoint dir %r" % new)
 
     def get_notebook_names(self, path=''):
         """List all notebook names in the notebook dir and path."""
         path = path.strip('/')
         if not os.path.isdir(self.get_os_path(path=path)):
             raise web.HTTPError(404, 'Directory not found: ' + path)
         names = glob.glob(self.get_os_path('*'+self.filename_ext, path))
         names = [os.path.basename(name)
                  for name in names]
         return names
 
     def increment_filename(self, basename, path='', ext='.ipynb'):
         """Return a non-used filename of the form basename<int>."""
         path = path.strip('/')
         for i in itertools.count():
             name = u'{basename}{i}{ext}'.format(basename=basename, i=i, ext=ext)
             os_path = self.get_os_path(name, path)
             if not os.path.isfile(os_path):
                 break
         return name
 
     def path_exists(self, path):
         """Does the API-style path (directory) actually exist?
         
         Parameters
         ----------
         path : string
             The path to check. This is an API path (`/` separated,
             relative to base notebook-dir).
         
         Returns
         -------
         exists : bool
             Whether the path is indeed a directory.
         """
         path = path.strip('/')
         os_path = self.get_os_path(path=path)
         return os.path.isdir(os_path)
     
     def get_os_path(self, name=None, path=''):
         """Given a notebook name and a URL path, return its file system
         path.
 
         Parameters
         ----------
         name : string
             The name of a notebook file with the .ipynb extension
         path : string
             The relative URL path (with '/' as separator) to the named
             notebook.
 
         Returns
         -------
         path : string
             A file system path that combines notebook_dir (location where
             server started), the relative path, and the filename with the
             current operating system's url.
         """
         parts = path.strip('/').split('/')
         parts = [p for p in parts if p != ''] # remove duplicate splits
         if name is not None:
             parts.append(name)
         path = os.path.join(self.notebook_dir, *parts)
         return path
 
     def notebook_exists(self, name, path=''):
         """Returns a True if the notebook exists. Else, returns False.
 
         Parameters
         ----------
         name : string
             The name of the notebook you are checking.
         path : string
             The relative path to the notebook (with '/' as separator)
 
         Returns
         -------
         bool
         """
         path = path.strip('/')
         nbpath = self.get_os_path(name, path=path)
         return os.path.isfile(nbpath)
 
     # TODO: Remove this after we create the contents web service and directories are
     # no longer listed by the notebook web service.
     def list_dirs(self, path):
         """List the directories for a given API style path."""
         path = path.strip('/')
         os_path = self.get_os_path('', path)
+        if not os.path.isdir(os_path):
+            raise web.HTTPError(404, u'diretory does not exist: %r' % os_path)
         dir_names = os.listdir(os_path)
         dirs = []
         for name in dir_names:
             os_path = self.get_os_path(name, path)
             if os.path.isdir(os_path) and not name.startswith('.'):
-                model = self.get_dir_model(name, path)
+                try:
+                    model = self.get_dir_model(name, path)
+                except IOError:
+                    pass
                 dirs.append(model)
         dirs = sorted(dirs, key=lambda item: item['name'])
         return dirs
 
     # TODO: Remove this after we create the contents web service and directories are
     # no longer listed by the notebook web service.
     def get_dir_model(self, name, path=''):
         """Get the directory model given a directory name and its API style path"""
         path = path.strip('/')
         os_path = self.get_os_path(name, path)
         if not os.path.isdir(os_path):
             raise IOError('directory does not exist: %r' % os_path)
         info = os.stat(os_path)
         last_modified = tz.utcfromtimestamp(info.st_mtime)
         created = tz.utcfromtimestamp(info.st_ctime)
         # Create the notebook model.
         model ={}
         model['name'] = name
         model['path'] = path
         model['last_modified'] = last_modified
         model['created'] = created
         model['type'] = 'directory'
         return model
 
     def list_notebooks(self, path):
         """Returns a list of dictionaries that are the standard model
         for all notebooks in the relative 'path'.
         
         Parameters
         ----------
         path : str
             the URL path that describes the relative path for the
             listed notebooks
         
         Returns
         -------
         notebooks : list of dicts
             a list of the notebook models without 'content'
         """
         path = path.strip('/')
         notebook_names = self.get_notebook_names(path)
-        index = []
-        notebooks = []
-        for name in notebook_names:
-            model = self.get_notebook_model(name, path, content=False)
-            if name.lower() == 'index.ipynb':
-                index.append(model)
-            else:
-                notebooks.append(model)
+        notebooks = [self.get_notebook_model(name, path, content=False) for name in notebook_names]
         notebooks = sorted(notebooks, key=lambda item: item['name'])
-        notebooks = index + self.list_dirs(path) + notebooks
         return notebooks
 
     def get_notebook_model(self, name, path='', content=True):
         """ Takes a path and name for a notebook and returns its model
         
         Parameters
         ----------
         name : str
             the name of the notebook
         path : str
             the URL path that describes the relative path for
             the notebook
             
         Returns
         -------
         model : dict
             the notebook model. If contents=True, returns the 'contents' 
             dict in the model as well.
         """
         path = path.strip('/')
         if not self.notebook_exists(name=name, path=path):
             raise web.HTTPError(404, u'Notebook does not exist: %s' % name)
         os_path = self.get_os_path(name, path)
         info = os.stat(os_path)
         last_modified = tz.utcfromtimestamp(info.st_mtime)
         created = tz.utcfromtimestamp(info.st_ctime)
         # Create the notebook model.
         model ={}
         model['name'] = name
         model['path'] = path
         model['last_modified'] = last_modified
         model['created'] = created
+        model['type'] = 'notebook'
         if content:
             with io.open(os_path, 'r', encoding='utf-8') as f:
                 try:
                     nb = current.read(f, u'json')
                 except Exception as e:
                     raise web.HTTPError(400, u"Unreadable Notebook: %s %s" % (os_path, e))
             self.mark_trusted_cells(nb, path, name)
             model['content'] = nb
         return model
 
     def save_notebook_model(self, model, name='', path=''):
         """Save the notebook model and return the model with no content."""
         path = path.strip('/')
 
         if 'content' not in model:
             raise web.HTTPError(400, u'No notebook JSON data provided')
-        
+
         # One checkpoint should always exist
         if self.notebook_exists(name, path) and not self.list_checkpoints(name, path):
             self.create_checkpoint(name, path)
 
         new_path = model.get('path', path).strip('/')
         new_name = model.get('name', name)
 
         if path != new_path or name != new_name:
             self.rename_notebook(name, path, new_name, new_path)
 
         # Save the notebook file
         os_path = self.get_os_path(new_name, new_path)
         nb = current.to_notebook_json(model['content'])
         
         self.check_and_sign(nb, new_path, new_name)
         
         if 'name' in nb['metadata']:
             nb['metadata']['name'] = u''
         try:
             self.log.debug("Autosaving notebook %s", os_path)
             with io.open(os_path, 'w', encoding='utf-8') as f:
                 current.write(nb, f, u'json')
         except Exception as e:
             raise web.HTTPError(400, u'Unexpected error while autosaving notebook: %s %s' % (os_path, e))
 
         # Save .py script as well
         if self.save_script:
             py_path = os.path.splitext(os_path)[0] + '.py'
             self.log.debug("Writing script %s", py_path)
             try:
                 with io.open(py_path, 'w', encoding='utf-8') as f:
                     current.write(nb, f, u'py')
             except Exception as e:
                 raise web.HTTPError(400, u'Unexpected error while saving notebook as script: %s %s' % (py_path, e))
 
         model = self.get_notebook_model(new_name, new_path, content=False)
         return model
 
     def update_notebook_model(self, model, name, path=''):
         """Update the notebook's path and/or name"""
         path = path.strip('/')
         new_name = model.get('name', name)
         new_path = model.get('path', path).strip('/')
         if path != new_path or name != new_name:
             self.rename_notebook(name, path, new_name, new_path)
         model = self.get_notebook_model(new_name, new_path, content=False)
         return model
 
     def delete_notebook_model(self, name, path=''):
         """Delete notebook by name and path."""
         path = path.strip('/')
         os_path = self.get_os_path(name, path)
         if not os.path.isfile(os_path):
             raise web.HTTPError(404, u'Notebook does not exist: %s' % os_path)
         
         # clear checkpoints
         for checkpoint in self.list_checkpoints(name, path):
             checkpoint_id = checkpoint['id']
             cp_path = self.get_checkpoint_path(checkpoint_id, name, path)
             if os.path.isfile(cp_path):
                 self.log.debug("Unlinking checkpoint %s", cp_path)
                 os.unlink(cp_path)
         
         self.log.debug("Unlinking notebook %s", os_path)
         os.unlink(os_path)
 
     def rename_notebook(self, old_name, old_path, new_name, new_path):
         """Rename a notebook."""
         old_path = old_path.strip('/')
         new_path = new_path.strip('/')
         if new_name == old_name and new_path == old_path:
             return
         
         new_os_path = self.get_os_path(new_name, new_path)
         old_os_path = self.get_os_path(old_name, old_path)
 
         # Should we proceed with the move?
         if os.path.isfile(new_os_path):
             raise web.HTTPError(409, u'Notebook with name already exists: %s' % new_os_path)
         if self.save_script:
             old_py_path = os.path.splitext(old_os_path)[0] + '.py'
             new_py_path = os.path.splitext(new_os_path)[0] + '.py'
             if os.path.isfile(new_py_path):
                 raise web.HTTPError(409, u'Python script with name already exists: %s' % new_py_path)
 
         # Move the notebook file
         try:
             os.rename(old_os_path, new_os_path)
         except Exception as e:
             raise web.HTTPError(500, u'Unknown error renaming notebook: %s %s' % (old_os_path, e))
 
         # Move the checkpoints
         old_checkpoints = self.list_checkpoints(old_name, old_path)
         for cp in old_checkpoints:
             checkpoint_id = cp['id']
             old_cp_path = self.get_checkpoint_path(checkpoint_id, old_name, old_path)
             new_cp_path = self.get_checkpoint_path(checkpoint_id, new_name, new_path)
             if os.path.isfile(old_cp_path):
                 self.log.debug("Renaming checkpoint %s -> %s", old_cp_path, new_cp_path)
                 os.rename(old_cp_path, new_cp_path)
 
         # Move the .py script
         if self.save_script:
             os.rename(old_py_path, new_py_path)
   
     # Checkpoint-related utilities
     
     def get_checkpoint_path(self, checkpoint_id, name, path=''):
         """find the path to a checkpoint"""
         path = path.strip('/')
         basename, _ = os.path.splitext(name)
         filename = u"{name}-{checkpoint_id}{ext}".format(
             name=basename,
             checkpoint_id=checkpoint_id,
             ext=self.filename_ext,
         )
         cp_path = os.path.join(path, self.checkpoint_dir, filename)
         return cp_path
 
     def get_checkpoint_model(self, checkpoint_id, name, path=''):
         """construct the info dict for a given checkpoint"""
         path = path.strip('/')
         cp_path = self.get_checkpoint_path(checkpoint_id, name, path)
         stats = os.stat(cp_path)
         last_modified = tz.utcfromtimestamp(stats.st_mtime)
         info = dict(
             id = checkpoint_id,
             last_modified = last_modified,
         )
         return info
         
     # public checkpoint API
     
     def create_checkpoint(self, name, path=''):
         """Create a checkpoint from the current state of a notebook"""
         path = path.strip('/')
         nb_path = self.get_os_path(name, path)
         # only the one checkpoint ID:
         checkpoint_id = u"checkpoint"
         cp_path = self.get_checkpoint_path(checkpoint_id, name, path)
         self.log.debug("creating checkpoint for notebook %s", name)
         if not os.path.exists(self.checkpoint_dir):
             os.mkdir(self.checkpoint_dir)
         shutil.copy2(nb_path, cp_path)
         
         # return the checkpoint info
         return self.get_checkpoint_model(checkpoint_id, name, path)
     
     def list_checkpoints(self, name, path=''):
         """list the checkpoints for a given notebook
         
         This notebook manager currently only supports one checkpoint per notebook.
         """
         path = path.strip('/')
         checkpoint_id = "checkpoint"
         path = self.get_checkpoint_path(checkpoint_id, name, path)
         if not os.path.exists(path):
             return []
         else:
             return [self.get_checkpoint_model(checkpoint_id, name, path)]
         
     
     def restore_checkpoint(self, checkpoint_id, name, path=''):
         """restore a notebook to a checkpointed state"""
         path = path.strip('/')
         self.log.info("restoring Notebook %s from checkpoint %s", name, checkpoint_id)
         nb_path = self.get_os_path(name, path)
         cp_path = self.get_checkpoint_path(checkpoint_id, name, path)
         if not os.path.isfile(cp_path):
             self.log.debug("checkpoint file does not exist: %s", cp_path)
             raise web.HTTPError(404,
                 u'Notebook checkpoint does not exist: %s-%s' % (name, checkpoint_id)
             )
         # ensure notebook is readable (never restore from an unreadable notebook)
         with io.open(cp_path, 'r', encoding='utf-8') as f:
             nb = current.read(f, u'json')
         shutil.copy2(cp_path, nb_path)
         self.log.debug("copying %s -> %s", cp_path, nb_path)
     
     def delete_checkpoint(self, checkpoint_id, name, path=''):
         """delete a notebook's checkpoint"""
         path = path.strip('/')
         cp_path = self.get_checkpoint_path(checkpoint_id, name, path)
         if not os.path.isfile(cp_path):
             raise web.HTTPError(404,
                 u'Notebook checkpoint does not exist: %s%s-%s' % (path, name, checkpoint_id)
             )
         self.log.debug("unlinking %s", cp_path)
         os.unlink(cp_path)
     
     def info_string(self):
         return "Serving notebooks from local directory: %s" % self.notebook_dir

IPython/html/services/notebooks/handlers.py

@@ -1,280 +1,290 b''
 """Tornado handlers for the notebooks web service.
 
 Authors:
 
 * Brian Granger
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import json
 
 from tornado import web
 
 from IPython.html.utils import url_path_join, url_escape
 from IPython.utils.jsonutil import date_default
 
 from IPython.html.base.handlers import (IPythonHandler, json_errors,
                                     notebook_path_regex, path_regex,
                                     notebook_name_regex)
 
 #-----------------------------------------------------------------------------
 # Notebook web service handlers
 #-----------------------------------------------------------------------------
 
 
 class NotebookHandler(IPythonHandler):
 
     SUPPORTED_METHODS = (u'GET', u'PUT', u'PATCH', u'POST', u'DELETE')
 
     def notebook_location(self, name, path=''):
         """Return the full URL location of a notebook based.
         
         Parameters
         ----------
         name : unicode
             The base name of the notebook, such as "foo.ipynb".
         path : unicode
             The URL path of the notebook.
         """
         return url_escape(url_path_join(
             self.base_project_url, 'api', 'notebooks', path, name
         ))
 
     def _finish_model(self, model, location=True):
         """Finish a JSON request with a model, setting relevant headers, etc."""
         if location:
             location = self.notebook_location(model['name'], model['path'])
             self.set_header('Location', location)
         self.set_header('Last-Modified', model['last_modified'])
         self.finish(json.dumps(model, default=date_default))
     
     @web.authenticated
     @json_errors
     def get(self, path='', name=None):
         """Return a Notebook or list of notebooks.
 
         * GET with path and no notebook name lists notebooks in a directory
         * GET with path and notebook name returns notebook JSON
         """
         nbm = self.notebook_manager
         # Check to see if a notebook name was given
         if name is None:
-            # List notebooks in 'path'
-            notebooks = nbm.list_notebooks(path)
+            # TODO: Remove this after we create the contents web service and directories are
+            # no longer listed by the notebook web service. This should only handle notebooks
+            # and not directories.
+            dirs = nbm.list_dirs(path)
+            notebooks = []
+            index = []
+            for nb in nbm.list_notebooks(path):
+                if nb['name'].lower() == 'index.ipynb':
+                    index.append(nb)
+                else:
+                    notebooks.append(nb)
+            notebooks = index + dirs + notebooks
             self.finish(json.dumps(notebooks, default=date_default))
             return
         # get and return notebook representation
         model = nbm.get_notebook_model(name, path)
         self._finish_model(model, location=False)
 
     @web.authenticated
     @json_errors
     def patch(self, path='', name=None):
         """PATCH renames a notebook without re-uploading content."""
         nbm = self.notebook_manager
         if name is None:
             raise web.HTTPError(400, u'Notebook name missing')
         model = self.get_json_body()
         if model is None:
             raise web.HTTPError(400, u'JSON body missing')
         model = nbm.update_notebook_model(model, name, path)
         self._finish_model(model)
     
     def _copy_notebook(self, copy_from, path, copy_to=None):
         """Copy a notebook in path, optionally specifying the new name.
         
         Only support copying within the same directory.
         """
         self.log.info(u"Copying notebook from %s/%s to %s/%s",
             path, copy_from,
             path, copy_to or '',
         )
         model = self.notebook_manager.copy_notebook(copy_from, copy_to, path)
         self.set_status(201)
         self._finish_model(model)
     
     def _upload_notebook(self, model, path, name=None):
         """Upload a notebook
         
         If name specified, create it in path/name.
         """
         self.log.info(u"Uploading notebook to %s/%s", path, name or '')
         if name:
             model['name'] = name
         
         model = self.notebook_manager.create_notebook_model(model, path)
         self.set_status(201)
         self._finish_model(model)
     
     def _create_empty_notebook(self, path, name=None):
         """Create an empty notebook in path
         
         If name specified, create it in path/name.
         """
         self.log.info(u"Creating new notebook in %s/%s", path, name or '')
         model = {}
         if name:
             model['name'] = name
         model = self.notebook_manager.create_notebook_model(model, path=path)
         self.set_status(201)
         self._finish_model(model)
     
     def _save_notebook(self, model, path, name):
         """Save an existing notebook."""
         self.log.info(u"Saving notebook at %s/%s", path, name)
         model = self.notebook_manager.save_notebook_model(model, name, path)
         if model['path'] != path.strip('/') or model['name'] != name:
             # a rename happened, set Location header
             location = True
         else:
             location = False
         self._finish_model(model, location)
     
     @web.authenticated
     @json_errors
     def post(self, path='', name=None):
         """Create a new notebook in the specified path.
         
         POST creates new notebooks. The server always decides on the notebook name.
         
         POST /api/notebooks/path
           New untitled notebook in path. If content specified, upload a
           notebook, otherwise start empty.
         POST /api/notebooks/path?copy=OtherNotebook.ipynb
           New copy of OtherNotebook in path
         """
         
         if name is not None:
             raise web.HTTPError(400, "Only POST to directories. Use PUT for full names.")
         
         model = self.get_json_body()
         
         if model is not None:
             copy_from = model.get('copy_from')
             if copy_from:
                 if model.get('content'):
                     raise web.HTTPError(400, "Can't upload and copy at the same time.")
                 self._copy_notebook(copy_from, path)
             else:
                 self._upload_notebook(model, path)
         else:
             self._create_empty_notebook(path)
 
     @web.authenticated
     @json_errors
     def put(self, path='', name=None):
         """Saves the notebook in the location specified by name and path.
         
         PUT is very similar to POST, but the requester specifies the name,
         whereas with POST, the server picks the name.
         
         PUT /api/notebooks/path/Name.ipynb
           Save notebook at ``path/Name.ipynb``. Notebook structure is specified
           in `content` key of JSON request body. If content is not specified,
           create a new empty notebook.
         PUT /api/notebooks/path/Name.ipynb?copy=OtherNotebook.ipynb
           Copy OtherNotebook to Name
         """
         if name is None:
             raise web.HTTPError(400, "Only PUT to full names. Use POST for directories.")
         
         model = self.get_json_body()
         if model:
             copy_from = model.get('copy_from')
             if copy_from:
                 if model.get('content'):
                     raise web.HTTPError(400, "Can't upload and copy at the same time.")
                 self._copy_notebook(copy_from, path, name)
             elif self.notebook_manager.notebook_exists(name, path):
                 self._save_notebook(model, path, name)
             else:
                 self._upload_notebook(model, path, name)
         else:
             self._create_empty_notebook(path, name)
 
     @web.authenticated
     @json_errors
     def delete(self, path='', name=None):
         """delete the notebook in the given notebook path"""
         nbm = self.notebook_manager
         nbm.delete_notebook_model(name, path)
         self.set_status(204)
         self.finish()
 
 
 class NotebookCheckpointsHandler(IPythonHandler):
     
     SUPPORTED_METHODS = ('GET', 'POST')
     
     @web.authenticated
     @json_errors
     def get(self, path='', name=None):
         """get lists checkpoints for a notebook"""
         nbm = self.notebook_manager
         checkpoints = nbm.list_checkpoints(name, path)
         data = json.dumps(checkpoints, default=date_default)
         self.finish(data)
     
     @web.authenticated
     @json_errors
     def post(self, path='', name=None):
         """post creates a new checkpoint"""
         nbm = self.notebook_manager
         checkpoint = nbm.create_checkpoint(name, path)
         data = json.dumps(checkpoint, default=date_default)
         location = url_path_join(self.base_project_url, 'api/notebooks',
             path, name, 'checkpoints', checkpoint['id'])
         self.set_header('Location', url_escape(location))
         self.set_status(201)
         self.finish(data)
 
 
 class ModifyNotebookCheckpointsHandler(IPythonHandler):
     
     SUPPORTED_METHODS = ('POST', 'DELETE')
     
     @web.authenticated
     @json_errors
     def post(self, path, name, checkpoint_id):
         """post restores a notebook from a checkpoint"""
         nbm = self.notebook_manager
         nbm.restore_checkpoint(checkpoint_id, name, path)
         self.set_status(204)
         self.finish()
     
     @web.authenticated
     @json_errors
     def delete(self, path, name, checkpoint_id):
         """delete clears a checkpoint for a given notebook"""
         nbm = self.notebook_manager
         nbm.delete_checkpoint(checkpoint_id, name, path)
         self.set_status(204)
         self.finish()
         
 #-----------------------------------------------------------------------------
 # URL to handler mappings
 #-----------------------------------------------------------------------------
 
 
 _checkpoint_id_regex = r"(?P<checkpoint_id>[\w-]+)"
 
 default_handlers = [
     (r"/api/notebooks%s/checkpoints" % notebook_path_regex, NotebookCheckpointsHandler),
     (r"/api/notebooks%s/checkpoints/%s" % (notebook_path_regex, _checkpoint_id_regex),
         ModifyNotebookCheckpointsHandler),
     (r"/api/notebooks%s" % notebook_path_regex, NotebookHandler),
     (r"/api/notebooks%s" % path_regex, NotebookHandler),
 ]
 
 
 

IPython/html/services/notebooks/nbmanager.py

@@ -1,198 +1,218 b''
 """A base class notebook manager.
 
 Authors:
 
 * Brian Granger
 * Zach Sailer
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import os
 
 from IPython.config.configurable import LoggingConfigurable
 from IPython.nbformat import current, sign
 from IPython.utils import py3compat
 from IPython.utils.traitlets import Instance, Unicode, TraitError
 
 #-----------------------------------------------------------------------------
 # Classes
 #-----------------------------------------------------------------------------
 
 class NotebookManager(LoggingConfigurable):
 
     # Todo:
     # The notebook_dir attribute is used to mean a couple of different things:
     # 1. Where the notebooks are stored if FileNotebookManager is used.
     # 2. The cwd of the kernel for a project.
     # Right now we use this attribute in a number of different places and
     # we are going to have to disentangle all of this.
     notebook_dir = Unicode(py3compat.getcwd(), config=True, help="""
             The directory to use for notebooks.
             """)
 
     filename_ext = Unicode(u'.ipynb')
     
     notary = Instance(sign.NotebookNotary)
     def _notary_default(self):
         return sign.NotebookNotary(parent=self)
     
     def check_and_sign(self, nb, path, name):
         """Check for trusted cells, and sign the notebook.
         
         Called as a part of saving notebooks.
         """
         if self.notary.check_cells(nb):
             self.notary.sign(nb)
         else:
             self.log.warn("Saving untrusted notebook %s/%s", path, name)
     
     def mark_trusted_cells(self, nb, path, name):
         """Mark cells as trusted if the notebook signature matches.
         
         Called as a part of loading notebooks.
         """
         trusted = self.notary.check_signature(nb)
         if not trusted:
             self.log.warn("Notebook %s/%s is not trusted", path, name)
         self.notary.mark_cells(nb, trusted)
     
     def path_exists(self, path):
         """Does the API-style path (directory) actually exist?
         
         Override this method in subclasses.
         
         Parameters
         ----------
         path : string
             The 
         
         Returns
         -------
         exists : bool
             Whether the path does indeed exist.
         """
         raise NotImplementedError
     
     def _notebook_dir_changed(self, name, old, new):
         """Do a bit of validation of the notebook dir."""
         if not os.path.isabs(new):
             # If we receive a non-absolute path, make it absolute.
             self.notebook_dir = os.path.abspath(new)
             return
         if os.path.exists(new) and not os.path.isdir(new):
             raise TraitError("notebook dir %r is not a directory" % new)
         if not os.path.exists(new):
             self.log.info("Creating notebook dir %s", new)
             try:
                 os.mkdir(new)
             except:
                 raise TraitError("Couldn't create notebook dir %r" % new)
 
     # Main notebook API
 
     def increment_filename(self, basename, path=''):
         """Increment a notebook filename without the .ipynb to make it unique.
         
         Parameters
         ----------
         basename : unicode
             The name of a notebook without the ``.ipynb`` file extension.
         path : unicode
             The URL path of the notebooks directory
         """
         return basename
 
+    # TODO: Remove this after we create the contents web service and directories are
+    # no longer listed by the notebook web service.
+    def list_dirs(self, path):
+        """List the directory models for a given API style path."""
+        raise NotImplementedError('must be implemented in a subclass')
+
+    # TODO: Remove this after we create the contents web service and directories are
+    # no longer listed by the notebook web service.
+    def get_dir_model(self, name, path=''):
+        """Get the directory model given a directory name and its API style path.
+        
+        The keys in the model should be:
+        * name
+        * path
+        * last_modified
+        * created
+        * type='directory'
+        """
+        raise NotImplementedError('must be implemented in a subclass')
+
     def list_notebooks(self, path=''):
         """Return a list of notebook dicts without content.
 
         This returns a list of dicts, each of the form::
 
             dict(notebook_id=notebook,name=name)
 
         This list of dicts should be sorted by name::
 
             data = sorted(data, key=lambda item: item['name'])
         """
         raise NotImplementedError('must be implemented in a subclass')
 
     def get_notebook_model(self, name, path='', content=True):
         """Get the notebook model with or without content."""
         raise NotImplementedError('must be implemented in a subclass')
 
     def save_notebook_model(self, model, name, path=''):
         """Save the notebook model and return the model with no content."""
         raise NotImplementedError('must be implemented in a subclass')
 
     def update_notebook_model(self, model, name, path=''):
         """Update the notebook model and return the model with no content."""
         raise NotImplementedError('must be implemented in a subclass')
 
     def delete_notebook_model(self, name, path=''):
         """Delete notebook by name and path."""
         raise NotImplementedError('must be implemented in a subclass')
 
     def create_notebook_model(self, model=None, path=''):
         """Create a new notebook and return its model with no content."""
         path = path.strip('/')
         if model is None:
             model = {}
         if 'content' not in model:
             metadata = current.new_metadata(name=u'')
             model['content'] = current.new_notebook(metadata=metadata)
         if 'name' not in model:
             model['name'] = self.increment_filename('Untitled', path)
             
         model['path'] = path
         model = self.save_notebook_model(model, model['name'], model['path'])
         return model
 
     def copy_notebook(self, from_name, to_name=None, path=''):
         """Copy an existing notebook and return its new model.
         
         If to_name not specified, increment `from_name-Copy#.ipynb`.
         """
         path = path.strip('/')
         model = self.get_notebook_model(from_name, path)
         if not to_name:
             base = os.path.splitext(from_name)[0] + '-Copy'
             to_name = self.increment_filename(base, path)
         model['name'] = to_name
         model = self.save_notebook_model(model, to_name, path)
         return model
     
     # Checkpoint-related
     
     def create_checkpoint(self, name, path=''):
         """Create a checkpoint of the current state of a notebook
         
         Returns a checkpoint_id for the new checkpoint.
         """
         raise NotImplementedError("must be implemented in a subclass")
     
     def list_checkpoints(self, name, path=''):
         """Return a list of checkpoints for a given notebook"""
         return []
     
     def restore_checkpoint(self, checkpoint_id, name, path=''):
         """Restore a notebook from one of its checkpoints"""
         raise NotImplementedError("must be implemented in a subclass")
 
     def delete_checkpoint(self, checkpoint_id, name, path=''):
         """delete a checkpoint for a notebook"""
         raise NotImplementedError("must be implemented in a subclass")
     
     def log_info(self):
         self.log.info(self.info_string())
     
     def info_string(self):
         return "Serving notebooks"

IPython/html/services/notebooks/tests/test_notebooks_api.py

@@ -1,329 +1,329 b''
 # coding: utf-8
 """Test the notebooks webservice API."""
 
 import io
 import json
 import os
 import shutil
 from unicodedata import normalize
 
 pjoin = os.path.join
 
 import requests
 
 from IPython.html.utils import url_path_join, url_escape
 from IPython.html.tests.launchnotebook import NotebookTestBase, assert_http_error
 from IPython.nbformat import current
 from IPython.nbformat.current import (new_notebook, write, read, new_worksheet,
                                       new_heading_cell, to_notebook_json)
 from IPython.nbformat import v2
 from IPython.utils import py3compat
 from IPython.utils.data import uniq_stable
 
 
 # TODO: Remove this after we create the contents web service and directories are
 # no longer listed by the notebook web service.
 def notebooks_only(nb_list):
-    return [nb for nb in nb_list if 'type' not in nb]
+    return [nb for nb in nb_list if nb['type']=='notebook']
 
 
 class NBAPI(object):
     """Wrapper for notebook API calls."""
     def __init__(self, base_url):
         self.base_url = base_url
 
     def _req(self, verb, path, body=None):
         response = requests.request(verb,
                 url_path_join(self.base_url, 'api/notebooks', path),
                 data=body,
         )
         response.raise_for_status()
         return response
 
     def list(self, path='/'):
         return self._req('GET', path)
 
     def read(self, name, path='/'):
         return self._req('GET', url_path_join(path, name))
 
     def create_untitled(self, path='/'):
         return self._req('POST', path)
 
     def upload_untitled(self, body, path='/'):
         return self._req('POST', path, body)
 
     def copy_untitled(self, copy_from, path='/'):
         body = json.dumps({'copy_from':copy_from})
         return self._req('POST', path, body)
 
     def create(self, name, path='/'):
         return self._req('PUT', url_path_join(path, name))
 
     def upload(self, name, body, path='/'):
         return self._req('PUT', url_path_join(path, name), body)
 
     def copy(self, copy_from, copy_to, path='/'):
         body = json.dumps({'copy_from':copy_from})
         return self._req('PUT', url_path_join(path, copy_to), body)
 
     def save(self, name, body, path='/'):
         return self._req('PUT', url_path_join(path, name), body)
 
     def delete(self, name, path='/'):
         return self._req('DELETE', url_path_join(path, name))
 
     def rename(self, name, path, new_name):
         body = json.dumps({'name': new_name})
         return self._req('PATCH', url_path_join(path, name), body)
 
     def get_checkpoints(self, name, path):
         return self._req('GET', url_path_join(path, name, 'checkpoints'))
 
     def new_checkpoint(self, name, path):
         return self._req('POST', url_path_join(path, name, 'checkpoints'))
 
     def restore_checkpoint(self, name, path, checkpoint_id):
         return self._req('POST', url_path_join(path, name, 'checkpoints', checkpoint_id))
 
     def delete_checkpoint(self, name, path, checkpoint_id):
         return self._req('DELETE', url_path_join(path, name, 'checkpoints', checkpoint_id))
 
 class APITest(NotebookTestBase):
     """Test the kernels web service API"""
     dirs_nbs = [('', 'inroot'),
                 ('Directory with spaces in', 'inspace'),
                 (u'unicodé', 'innonascii'),
                 ('foo', 'a'),
                 ('foo', 'b'),
                 ('foo', 'name with spaces'),
                 ('foo', u'unicodé'),
                 ('foo/bar', 'baz'),
                 (u'å b', u'ç d')
                ]
 
     dirs = uniq_stable([d for (d,n) in dirs_nbs])
     del dirs[0]  # remove ''
 
     def setUp(self):
         nbdir = self.notebook_dir.name
 
         for d in self.dirs:
             d.replace('/', os.sep)
             if not os.path.isdir(pjoin(nbdir, d)):
                 os.mkdir(pjoin(nbdir, d))
 
         for d, name in self.dirs_nbs:
             d = d.replace('/', os.sep)
             with io.open(pjoin(nbdir, d, '%s.ipynb' % name), 'w',
                          encoding='utf-8') as f:
                 nb = new_notebook(name=name)
                 write(nb, f, format='ipynb')
 
         self.nb_api = NBAPI(self.base_url())
 
     def tearDown(self):
         nbdir = self.notebook_dir.name
 
         for dname in ['foo', 'Directory with spaces in', u'unicodé', u'å b']:
             shutil.rmtree(pjoin(nbdir, dname), ignore_errors=True)
 
         if os.path.isfile(pjoin(nbdir, 'inroot.ipynb')):
             os.unlink(pjoin(nbdir, 'inroot.ipynb'))
 
     def test_list_notebooks(self):
         nbs = notebooks_only(self.nb_api.list().json())
         self.assertEqual(len(nbs), 1)
         self.assertEqual(nbs[0]['name'], 'inroot.ipynb')
 
         nbs = notebooks_only(self.nb_api.list('/Directory with spaces in/').json())
         self.assertEqual(len(nbs), 1)
         self.assertEqual(nbs[0]['name'], 'inspace.ipynb')
 
         nbs = notebooks_only(self.nb_api.list(u'/unicodé/').json())
         self.assertEqual(len(nbs), 1)
         self.assertEqual(nbs[0]['name'], 'innonascii.ipynb')
         self.assertEqual(nbs[0]['path'], u'unicodé')
 
         nbs = notebooks_only(self.nb_api.list('/foo/bar/').json())
         self.assertEqual(len(nbs), 1)
         self.assertEqual(nbs[0]['name'], 'baz.ipynb')
         self.assertEqual(nbs[0]['path'], 'foo/bar')
 
         nbs = notebooks_only(self.nb_api.list('foo').json())
         self.assertEqual(len(nbs), 4)
         nbnames = { normalize('NFC', n['name']) for n in nbs }
         expected = [ u'a.ipynb', u'b.ipynb', u'name with spaces.ipynb', u'unicodé.ipynb']
         expected = { normalize('NFC', name) for name in expected }
         self.assertEqual(nbnames, expected)
 
     def test_list_nonexistant_dir(self):
         with assert_http_error(404):
             self.nb_api.list('nonexistant')
 
     def test_get_contents(self):
         for d, name in self.dirs_nbs:
             nb = self.nb_api.read('%s.ipynb' % name, d+'/').json()
             self.assertEqual(nb['name'], u'%s.ipynb' % name)
             self.assertIn('content', nb)
             self.assertIn('metadata', nb['content'])
             self.assertIsInstance(nb['content']['metadata'], dict)
 
         # Name that doesn't exist - should be a 404
         with assert_http_error(404):
             self.nb_api.read('q.ipynb', 'foo')
 
     def _check_nb_created(self, resp, name, path):
         self.assertEqual(resp.status_code, 201)
         location_header = py3compat.str_to_unicode(resp.headers['Location'])
         self.assertEqual(location_header, url_escape(url_path_join(u'/api/notebooks', path, name)))
         self.assertEqual(resp.json()['name'], name)
         assert os.path.isfile(pjoin(
             self.notebook_dir.name,
             path.replace('/', os.sep),
             name,
         ))
 
     def test_create_untitled(self):
         resp = self.nb_api.create_untitled(path=u'å b')
         self._check_nb_created(resp, 'Untitled0.ipynb', u'å b')
 
         # Second time
         resp = self.nb_api.create_untitled(path=u'å b')
         self._check_nb_created(resp, 'Untitled1.ipynb', u'å b')
 
         # And two directories down
         resp = self.nb_api.create_untitled(path='foo/bar')
         self._check_nb_created(resp, 'Untitled0.ipynb', 'foo/bar')
 
     def test_upload_untitled(self):
         nb = new_notebook(name='Upload test')
         nbmodel = {'content': nb}
         resp = self.nb_api.upload_untitled(path=u'å b',
                                               body=json.dumps(nbmodel))
         self._check_nb_created(resp, 'Untitled0.ipynb', u'å b')
 
     def test_upload(self):
         nb = new_notebook(name=u'ignored')
         nbmodel = {'content': nb}
         resp = self.nb_api.upload(u'Upload tést.ipynb', path=u'å b',
                                               body=json.dumps(nbmodel))
         self._check_nb_created(resp, u'Upload tést.ipynb', u'å b')
 
     def test_upload_v2(self):
         nb = v2.new_notebook()
         ws = v2.new_worksheet()
         nb.worksheets.append(ws)
         ws.cells.append(v2.new_code_cell(input='print("hi")'))
         nbmodel = {'content': nb}
         resp = self.nb_api.upload(u'Upload tést.ipynb', path=u'å b',
                                               body=json.dumps(nbmodel))
         self._check_nb_created(resp, u'Upload tést.ipynb', u'å b')
         resp = self.nb_api.read(u'Upload tést.ipynb', u'å b')
         data = resp.json()
         self.assertEqual(data['content']['nbformat'], current.nbformat)
         self.assertEqual(data['content']['orig_nbformat'], 2)
 
     def test_copy_untitled(self):
         resp = self.nb_api.copy_untitled(u'ç d.ipynb', path=u'å b')
         self._check_nb_created(resp, u'ç d-Copy0.ipynb', u'å b')
 
     def test_copy(self):
         resp = self.nb_api.copy(u'ç d.ipynb', u'cøpy.ipynb', path=u'å b')
         self._check_nb_created(resp, u'cøpy.ipynb', u'å b')
 
     def test_delete(self):
         for d, name in self.dirs_nbs:
             resp = self.nb_api.delete('%s.ipynb' % name, d)
             self.assertEqual(resp.status_code, 204)
 
         for d in self.dirs + ['/']:
             nbs = notebooks_only(self.nb_api.list(d).json())
             self.assertEqual(len(nbs), 0)
 
     def test_rename(self):
         resp = self.nb_api.rename('a.ipynb', 'foo', 'z.ipynb')
         self.assertEqual(resp.headers['Location'].split('/')[-1], 'z.ipynb')
         self.assertEqual(resp.json()['name'], 'z.ipynb')
         assert os.path.isfile(pjoin(self.notebook_dir.name, 'foo', 'z.ipynb'))
 
         nbs = notebooks_only(self.nb_api.list('foo').json())
         nbnames = set(n['name'] for n in nbs)
         self.assertIn('z.ipynb', nbnames)
         self.assertNotIn('a.ipynb', nbnames)
 
     def test_rename_existing(self):
         with assert_http_error(409):
             self.nb_api.rename('a.ipynb', 'foo', 'b.ipynb')
 
     def test_save(self):
         resp = self.nb_api.read('a.ipynb', 'foo')
         nbcontent = json.loads(resp.text)['content']
         nb = to_notebook_json(nbcontent)
         ws = new_worksheet()
         nb.worksheets = [ws]
         ws.cells.append(new_heading_cell(u'Created by test ³'))
 
         nbmodel= {'name': 'a.ipynb', 'path':'foo', 'content': nb}
         resp = self.nb_api.save('a.ipynb', path='foo', body=json.dumps(nbmodel))
 
         nbfile = pjoin(self.notebook_dir.name, 'foo', 'a.ipynb')
         with io.open(nbfile, 'r', encoding='utf-8') as f:
             newnb = read(f, format='ipynb')
         self.assertEqual(newnb.worksheets[0].cells[0].source,
                          u'Created by test ³')
         nbcontent = self.nb_api.read('a.ipynb', 'foo').json()['content']
         newnb = to_notebook_json(nbcontent)
         self.assertEqual(newnb.worksheets[0].cells[0].source,
                          u'Created by test ³')
 
         # Save and rename
         nbmodel= {'name': 'a2.ipynb', 'path':'foo/bar', 'content': nb}
         resp = self.nb_api.save('a.ipynb', path='foo', body=json.dumps(nbmodel))
         saved = resp.json()
         self.assertEqual(saved['name'], 'a2.ipynb')
         self.assertEqual(saved['path'], 'foo/bar')
         assert os.path.isfile(pjoin(self.notebook_dir.name,'foo','bar','a2.ipynb'))
         assert not os.path.isfile(pjoin(self.notebook_dir.name, 'foo', 'a.ipynb'))
         with assert_http_error(404):
             self.nb_api.read('a.ipynb', 'foo')
 
     def test_checkpoints(self):
         resp = self.nb_api.read('a.ipynb', 'foo')
         r = self.nb_api.new_checkpoint('a.ipynb', 'foo')
         self.assertEqual(r.status_code, 201)
         cp1 = r.json()
         self.assertEqual(set(cp1), {'id', 'last_modified'})
         self.assertEqual(r.headers['Location'].split('/')[-1], cp1['id'])
 
         # Modify it
         nbcontent = json.loads(resp.text)['content']
         nb = to_notebook_json(nbcontent)
         ws = new_worksheet()
         nb.worksheets = [ws]
         hcell = new_heading_cell('Created by test')
         ws.cells.append(hcell)
         # Save
         nbmodel= {'name': 'a.ipynb', 'path':'foo', 'content': nb}
         resp = self.nb_api.save('a.ipynb', path='foo', body=json.dumps(nbmodel))
 
         # List checkpoints
         cps = self.nb_api.get_checkpoints('a.ipynb', 'foo').json()
         self.assertEqual(cps, [cp1])
 
         nbcontent = self.nb_api.read('a.ipynb', 'foo').json()['content']
         nb = to_notebook_json(nbcontent)
         self.assertEqual(nb.worksheets[0].cells[0].source, 'Created by test')
 
         # Restore cp1
         r = self.nb_api.restore_checkpoint('a.ipynb', 'foo', cp1['id'])
         self.assertEqual(r.status_code, 204)
         nbcontent = self.nb_api.read('a.ipynb', 'foo').json()['content']
         nb = to_notebook_json(nbcontent)
         self.assertEqual(nb.worksheets, [])
 
         # Delete cp1
         r = self.nb_api.delete_checkpoint('a.ipynb', 'foo', cp1['id'])
         self.assertEqual(r.status_code, 204)
         cps = self.nb_api.get_checkpoints('a.ipynb', 'foo').json()
         self.assertEqual(cps, [])