Create AsyncGenericFileCheckpoints

Author: Mariko Wakabayashi

jupyter_server/services/contents/filecheckpoints.py

@@ -9,6 +9,7 @@
 from .checkpoints import (
     AsyncCheckpoints,
     Checkpoints,
+    AsyncGenericCheckpointsMixin,
     GenericCheckpointsMixin,
 )
 from .fileio import AsyncFileManagerMixin, FileManagerMixin
@@ -266,3 +267,68 @@ def get_file_checkpoint(self, checkpoint_id, path):
             'content': content,
             'format': format,
         }
+
+
+class AsyncGenericFileCheckpoints(AsyncGenericCheckpointsMixin, AsyncFileCheckpoints):
+    """
+    Asynchronous Local filesystem Checkpoints that works with any conforming
+    ContentsManager.
+    """
+    async def create_file_checkpoint(self, content, format, path):
+        """Create a checkpoint from the current content of a file."""
+        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():
+            await self._save_file(os_checkpoint_path, content, format=format)
+
+        # return the checkpoint info
+        return await self.checkpoint_model(checkpoint_id, os_checkpoint_path)
+
+    async 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():
+            await self._save_notebook(os_checkpoint_path, nb)
+
+        # return the checkpoint info
+        return await self.checkpoint_model(checkpoint_id, os_checkpoint_path)
+
+    async def get_notebook_checkpoint(self, checkpoint_id, path):
+        """Get a checkpoint for a notebook."""
+        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)
+
+        return {
+            'type': 'notebook',
+            'content': await self._read_notebook(
+                os_checkpoint_path,
+                as_version=4,
+            ),
+        }
+
+    async def get_file_checkpoint(self, checkpoint_id, path):
+        """Get a checkpoint for a file."""
+        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)
+
+        content, format = await self._read_file(os_checkpoint_path, format=None)
+        return {
+            'type': 'file',
+            'content': content,
+            'format': format,
+        }

tests/services/contents/test_config.py

@@ -1,17 +1,18 @@
 import pytest
 
 from jupyter_server.services.contents.checkpoints import AsyncCheckpoints
-from jupyter_server.services.contents.filecheckpoints import GenericFileCheckpoints
+from jupyter_server.services.contents.filecheckpoints import AsyncGenericFileCheckpoints, GenericFileCheckpoints
 from jupyter_server.services.contents.manager import AsyncContentsManager
 
 
-@pytest.fixture
-def jp_server_config():
-    return {'FileContentsManager': {'checkpoints_class': GenericFileCheckpoints}}
+@pytest.fixture(params=[AsyncGenericFileCheckpoints, GenericFileCheckpoints])
+def jp_server_config(request):
+    return {'FileContentsManager': {'checkpoints_class': request.param}}
 
 
-def test_config_did_something(jp_serverapp):
-    assert isinstance(jp_serverapp.contents_manager.checkpoints, GenericFileCheckpoints)
+def test_config_did_something(jp_server_config, jp_serverapp):
+    assert isinstance(jp_serverapp.contents_manager.checkpoints,
+                      jp_server_config['FileContentsManager']['checkpoints_class'])
 
 
 async def test_async_contents_manager(jp_configurable_serverapp):