DEV: Allow CheckpointManagers to optimize for shared backends.

On `create_checkpoint` and `restore_checkpoint`, pass a path and a
`ContentsManager` to `CheckpointManager` instead of an already-loaded
model.

The `CheckpointManager` base class provides a correct implementation of
these methods that's generic across any ContentsManager, but subclasses
are free to specialize when the storage backend of `ContentsManager` is
shared.

IPython/html/services/contents/filemanager.py

@@ -222,59 +222,48 b' class FileCheckpointManager(FileManagerMixin, CheckpointManager):'
         except AttributeError:
             return getcwd()
 
-    # public checkpoint API
-    def create_file_checkpoint(self, content, format, path):
-        """Create a checkpoint from the current content of a notebook."""
-        path = path.strip('/')
-        # only the one checkpoint ID:
-        checkpoint_id = u"checkpoint"
-        os_checkpoint_path = self.checkpoint_path(checkpoint_id, path)
-        self.log.debug("creating checkpoint for %s", path)
-        with self.perm_to_403():
-            self._save_file(os_checkpoint_path, content, format=format)
-
-        # return the checkpoint info
-        return self.checkpoint_model(checkpoint_id, os_checkpoint_path)
-
-    def create_notebook_checkpoint(self, nb, path):
-        """Create a checkpoint from the current content of a notebook."""
-        path = path.strip('/')
-        # only the one checkpoint ID:
-        checkpoint_id = u"checkpoint"
-        os_checkpoint_path = self.checkpoint_path(checkpoint_id, path)
-        self.log.debug("creating checkpoint for %s", path)
-        with self.perm_to_403():
-            self._save_notebook(os_checkpoint_path, nb)
+    # ContentsManager-dependent checkpoint API
+    def create_checkpoint(self, contents_mgr, path):
+        """
+        Create a checkpoint.
 
-        # return the checkpoint info
-        return self.checkpoint_model(checkpoint_id, os_checkpoint_path)
+        If contents_mgr is backed by the local filesystem, just copy the
+        appropriate file to the checkpoint directory.  Otherwise, ask the
+        ContentsManager for a model and write it ourselves.
+        """
+        if contents_mgr.backend == 'local_file':
+            # We know that the file is in the local filesystem, so just copy
+            # from the base location to our location.
+            checkpoint_id = u'checkpoint'
+            src_path = contents_mgr._get_os_path(path)
+            dest_path = self.checkpoint_path(checkpoint_id, path)
+            self._copy(src_path, dest_path)
+            return self.checkpoint_model(checkpoint_id, dest_path)
+        else:
+            return super(FileCheckpointManager, self).create_checkpoint(
+                contents_mgr, path,
+            )
 
-    def get_checkpoint(self, checkpoint_id, path, type):
-        """Get the content of a checkpoint.
+    def restore_checkpoint(self, contents_mgr, checkpoint_id, path):
+        """
+        Restore a checkpoint.
 
-        Returns a model suitable for passing to ContentsManager.save.
+        If contents_mgr is backed by the local filesystem, just copy the
+        appropriate file from the checkpoint directory.  Otherwise, load the
+        model and pass it to ContentsManager.save.
         """
-        path = path.strip('/')
-        self.log.info("restoring %s from checkpoint %s", path, checkpoint_id)
-        os_checkpoint_path = self.checkpoint_path(checkpoint_id, path)
-        if not os.path.isfile(os_checkpoint_path):
-            self.no_such_checkpoint(path, checkpoint_id)
-        if type == 'notebook':
-            return {
-                'type': type,
-                'content': self._read_notebook(
-                    os_checkpoint_path,
-                    as_version=4,
-                ),
-            }
+        if contents_mgr.backend == 'local_file':
+            # We know that the file is in the local filesystem, so just copy
+            # from our base location to the location expected by content
+            src_path = self.checkpoint_path(checkpoint_id, path)
+            dest_path = contents_mgr._get_os_path(path)
+            self._copy(src_path, dest_path)
         else:
-            content, format = self._read_file(os_checkpoint_path, format=None)
-            return {
-                'type': type,
-                'content': content,
-                'format': format,
-            }
+            super(FileCheckpointManager, self).restore_checkpoint(
+                contents_mgr, checkpoint_id, path
+            )
 
+    # ContentsManager-independent checkpoint API
     def rename_checkpoint(self, checkpoint_id, old_path, new_path):
         """Rename a checkpoint from old_path to new_path."""
         old_cp_path = self.checkpoint_path(checkpoint_id, old_path)
@@ -341,6 +330,64 b' class FileCheckpointManager(FileManagerMixin, CheckpointManager):'
         )
         return info
 
+    def create_file_checkpoint(self, content, format, path):
+        """Create a checkpoint from the current content of a notebook."""
+        path = path.strip('/')
+        # only the one checkpoint ID:
+        checkpoint_id = u"checkpoint"
+        os_checkpoint_path = self.checkpoint_path(checkpoint_id, path)
+        self.log.debug("creating checkpoint for %s", path)
+        with self.perm_to_403():
+            self._save_file(os_checkpoint_path, content, format=format)
+
+        # return the checkpoint info
+        return self.checkpoint_model(checkpoint_id, os_checkpoint_path)
+
+    def create_notebook_checkpoint(self, nb, path):
+        """Create a checkpoint from the current content of a notebook."""
+        path = path.strip('/')
+        # only the one checkpoint ID:
+        checkpoint_id = u"checkpoint"
+        os_checkpoint_path = self.checkpoint_path(checkpoint_id, path)
+        self.log.debug("creating checkpoint for %s", path)
+        with self.perm_to_403():
+            self._save_notebook(os_checkpoint_path, nb)
+
+        # return the checkpoint info
+        return self.checkpoint_model(checkpoint_id, os_checkpoint_path)
+
+    def get_checkpoint(self, checkpoint_id, path, type):
+        """Get the content of a checkpoint.
+
+        Returns a model suitable for passing to ContentsManager.save.
+        """
+        path = path.strip('/')
+        self.log.info("restoring %s from checkpoint %s", path, checkpoint_id)
+        os_checkpoint_path = self.checkpoint_path(checkpoint_id, path)
+        if not os.path.isfile(os_checkpoint_path):
+            self.no_such_checkpoint(path, checkpoint_id)
+
+        if type == 'notebook':
+            return {
+                'type': type,
+                'content': self._read_notebook(
+                    os_checkpoint_path,
+                    as_version=4,
+                ),
+            }
+        elif type == 'file':
+            content, format = self._read_file(os_checkpoint_path, format=None)
+            return {
+                'type': type,
+                'content': content,
+                'format': format,
+            }
+        else:
+            raise web.HTTPError(
+                500,
+                u'Unexpected type %s' % type
+            )
+
     # Error Handling
     def no_such_checkpoint(self, path, checkpoint_id):
         raise web.HTTPError(
@@ -421,6 +468,9 b' class FileContentsManager(FileManagerMixin, ContentsManager):'
     def _checkpoint_manager_class_default(self):
         return FileCheckpointManager
 
+    def _backend_default(self):
+        return 'local_file'
+
     def is_hidden(self, path):
         """Does the API style path correspond to a hidden directory or file?
 
@@ -681,10 +731,7 b' class FileContentsManager(FileManagerMixin, ContentsManager):'
                 self._save_notebook(os_path, nb)
                 # One checkpoint should always exist for notebooks.
                 if not self.checkpoint_manager.list_checkpoints(path):
-                    self.checkpoint_manager.create_notebook_checkpoint(
-                        nb,
-                        path,
-                    )
+                    self.create_checkpoint(path)
             elif model['type'] == 'file':
                 # Missing format will be handled internally by _save_file.
                 self._save_file(os_path, model['content'], model.get('format'))

IPython/html/services/contents/manager.py

@@ -11,7 +11,6 b' import re'
 
 from tornado.web import HTTPError
 
-from IPython import nbformat
 from IPython.config.configurable import LoggingConfigurable
 from IPython.nbformat import sign, validate, ValidationError
 from IPython.nbformat.v4 import new_notebook
@@ -34,6 +33,28 b' class CheckpointManager(LoggingConfigurable):'
     """
     Base class for managing checkpoints for a ContentsManager.
     """
+
+    def create_checkpoint(self, contents_mgr, path):
+        model = contents_mgr.get(path, content=True)
+        type = model['type']
+        if type == 'notebook':
+            return self.create_notebook_checkpoint(
+                model['content'],
+                path,
+            )
+        elif type == 'file':
+            return self.create_file_checkpoint(
+                model['content'],
+                model['format'],
+                path,
+            )
+
+    def restore_checkpoint(self, contents_mgr, checkpoint_id, path):
+        """Restore a checkpoint."""
+        type = contents_mgr.get(path, content=False)['type']
+        model = self.get_checkpoint(checkpoint_id, path, type)
+        contents_mgr.save(model, path)
+
     def create_file_checkpoint(self, content, format, path):
         """Create a checkpoint of the current state of a file
 
@@ -159,6 +180,7 b' class ContentsManager(LoggingConfigurable):'
     checkpoint_manager_class = Type(CheckpointManager, config=True)
     checkpoint_manager = Instance(CheckpointManager, config=True)
     checkpoint_manager_kwargs = Dict(allow_none=False, config=True)
+    backend = Unicode(default_value="")
 
     def _checkpoint_manager_default(self):
         return self.checkpoint_manager_class(**self.checkpoint_manager_kwargs)
@@ -502,35 +524,16 b' class ContentsManager(LoggingConfigurable):'
     # Part 3: Checkpoints API
     def create_checkpoint(self, path):
         """Create a checkpoint."""
-        model = self.get(path, content=True)
-        type = model['type']
-        if type == 'notebook':
-            return self.checkpoint_manager.create_notebook_checkpoint(
-                model['content'],
-                path,
-            )
-        elif type == 'file':
-            return self.checkpoint_manager.create_file_checkpoint(
-                model['content'],
-                model['format'],
-                path,
-            )
-
-    def list_checkpoints(self, path):
-        return self.checkpoint_manager.list_checkpoints(path)
+        return self.checkpoint_manager.create_checkpoint(self, path)
 
     def restore_checkpoint(self, checkpoint_id, path):
         """
         Restore a checkpoint.
         """
-        return self.save(
-            model=self.checkpoint_manager.get_checkpoint(
-                checkpoint_id,
-                path,
-                self.get(path, content=False)['type']
-            ),
-            path=path,
-        )
+        self.checkpoint_manager.restore_checkpoint(self, checkpoint_id, path)
+
+    def list_checkpoints(self, path):
+        return self.checkpoint_manager.list_checkpoints(path)
 
     def delete_checkpoint(self, checkpoint_id, path):
         return self.checkpoint_manager.delete_checkpoint(checkpoint_id, path)

IPython/html/services/contents/tests/test_contents_api.py

@@ -614,3 +614,25 b' class APITest(NotebookTestBase):'
         with TemporaryDirectory() as td:
             with self.patch_cp_root(td):
                 self.test_file_checkpoints()
+
+    @contextmanager
+    def patch_cm_backend(self):
+        """
+        Temporarily patch our ContentsManager to present a different backend.
+        """
+        mgr = self.notebook.contents_manager
+        old_backend = mgr.backend
+        mgr.backend = ""
+        try:
+            yield
+        finally:
+            mgr.backend = old_backend
+
+    def test_checkpoints_empty_backend(self):
+        with self.patch_cm_backend():
+            self.test_checkpoints()
+
+        with self.patch_cm_backend():
+            self.test_file_checkpoints()
+
+