/usr/lib/python2.7/dist-packages/notebook/services/contents/checkpoints.py is in python-notebook 5.2.2-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 | """
Classes for managing Checkpoints.
"""
# Copyright (c) Jupyter Development Team.
# Distributed under the terms of the Modified BSD License.
from tornado.web import HTTPError
from traitlets.config.configurable import LoggingConfigurable
class Checkpoints(LoggingConfigurable):
"""
Base class for managing checkpoints for a ContentsManager.
Subclasses are required to implement:
create_checkpoint(self, contents_mgr, path)
restore_checkpoint(self, contents_mgr, checkpoint_id, path)
rename_checkpoint(self, checkpoint_id, old_path, new_path)
delete_checkpoint(self, checkpoint_id, path)
list_checkpoints(self, path)
"""
def create_checkpoint(self, contents_mgr, path):
"""Create a checkpoint."""
raise NotImplementedError("must be implemented in a subclass")
def restore_checkpoint(self, contents_mgr, checkpoint_id, path):
"""Restore a checkpoint"""
raise NotImplementedError("must be implemented in a subclass")
def rename_checkpoint(self, checkpoint_id, old_path, new_path):
"""Rename a single checkpoint from old_path to new_path."""
raise NotImplementedError("must be implemented in a subclass")
def delete_checkpoint(self, checkpoint_id, path):
"""delete a checkpoint for a file"""
raise NotImplementedError("must be implemented in a subclass")
def list_checkpoints(self, path):
"""Return a list of checkpoints for a given file"""
raise NotImplementedError("must be implemented in a subclass")
def rename_all_checkpoints(self, old_path, new_path):
"""Rename all checkpoints for old_path to new_path."""
for cp in self.list_checkpoints(old_path):
self.rename_checkpoint(cp['id'], old_path, new_path)
def delete_all_checkpoints(self, path):
"""Delete all checkpoints for the given path."""
for checkpoint in self.list_checkpoints(path):
self.delete_checkpoint(checkpoint['id'], path)
class GenericCheckpointsMixin(object):
"""
Helper for creating Checkpoints subclasses that can be used with any
ContentsManager.
Provides a ContentsManager-agnostic implementation of `create_checkpoint`
and `restore_checkpoint` in terms of the following operations:
- create_file_checkpoint(self, content, format, path)
- create_notebook_checkpoint(self, nb, path)
- get_file_checkpoint(self, checkpoint_id, path)
- get_notebook_checkpoint(self, checkpoint_id, path)
To create a generic CheckpointManager, add this mixin to a class that
implement the above four methods plus the remaining Checkpoints API
methods:
- delete_checkpoint(self, checkpoint_id, path)
- list_checkpoints(self, path)
- rename_checkpoint(self, checkpoint_id, old_path, new_path)
"""
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,
)
else:
raise HTTPError(500, u'Unexpected type %s' % type)
def restore_checkpoint(self, contents_mgr, checkpoint_id, path):
"""Restore a checkpoint."""
type = contents_mgr.get(path, content=False)['type']
if type == 'notebook':
model = self.get_notebook_checkpoint(checkpoint_id, path)
elif type == 'file':
model = self.get_file_checkpoint(checkpoint_id, path)
else:
raise HTTPError(500, u'Unexpected type %s' % type)
contents_mgr.save(model, path)
# Required Methods
def create_file_checkpoint(self, content, format, path):
"""Create a checkpoint of the current state of a file
Returns a checkpoint model for the new checkpoint.
"""
raise NotImplementedError("must be implemented in a subclass")
def create_notebook_checkpoint(self, nb, path):
"""Create a checkpoint of the current state of a file
Returns a checkpoint model for the new checkpoint.
"""
raise NotImplementedError("must be implemented in a subclass")
def get_file_checkpoint(self, checkpoint_id, path):
"""Get the content of a checkpoint for a non-notebook file.
Returns a dict of the form:
{
'type': 'file',
'content': <str>,
'format': {'text','base64'},
}
"""
raise NotImplementedError("must be implemented in a subclass")
def get_notebook_checkpoint(self, checkpoint_id, path):
"""Get the content of a checkpoint for a notebook.
Returns a dict of the form:
{
'type': 'notebook',
'content': <output of nbformat.read>,
}
"""
raise NotImplementedError("must be implemented in a subclass")
|