manager.py
346 lines
| 11.1 KiB
| text/x-python
|
PythonLexer
|
r17524 | """A base class for contents managers.""" | ||
|
r8180 | |||
|
r17524 | # Copyright (c) IPython Development Team. | ||
# Distributed under the terms of the Modified BSD License. | ||||
|
r8180 | |||
|
r15525 | from fnmatch import fnmatch | ||
|
r15292 | import itertools | ||
|
r18249 | import json | ||
|
r8181 | import os | ||
|
r8180 | |||
|
r17530 | from tornado.web import HTTPError | ||
|
r8180 | from IPython.config.configurable import LoggingConfigurable | ||
|
r14857 | from IPython.nbformat import current, sign | ||
|
r15525 | from IPython.utils.traitlets import Instance, Unicode, List | ||
|
r8180 | |||
|
r17524 | class ContentsManager(LoggingConfigurable): | ||
|
r17535 | """Base class for serving files and directories. | ||
This serves any text or binary file, | ||||
as well as directories, | ||||
with special handling for JSON notebook documents. | ||||
Most APIs take a path argument, | ||||
which is always an API-style unicode path, | ||||
and always refers to a directory. | ||||
- unicode, not url-escaped | ||||
- '/'-separated | ||||
- leading and trailing '/' will be stripped | ||||
- if unspecified, path defaults to '', | ||||
indicating the root path. | ||||
name is also unicode, and refers to a specfic target: | ||||
- unicode, not url-escaped | ||||
- must not contain '/' | ||||
- It refers to an individual filename | ||||
- It may refer to a directory name, | ||||
in the case of listing or creating directories. | ||||
""" | ||||
|
r17523 | |||
|
r14857 | notary = Instance(sign.NotebookNotary) | ||
def _notary_default(self): | ||||
return sign.NotebookNotary(parent=self) | ||||
|
r17523 | |||
|
r17529 | hide_globs = List(Unicode, [ | ||
u'__pycache__', '*.pyc', '*.pyo', | ||||
'.DS_Store', '*.so', '*.dylib', '*~', | ||||
], config=True, help=""" | ||||
|
r15525 | Glob patterns to hide in file and directory listings. | ||
""") | ||||
|
r17535 | untitled_notebook = Unicode("Untitled", config=True, | ||
help="The base name used when creating untitled notebooks." | ||||
) | ||||
untitled_file = Unicode("untitled", config=True, | ||||
help="The base name used when creating untitled files." | ||||
) | ||||
untitled_directory = Unicode("Untitled Folder", config=True, | ||||
help="The base name used when creating untitled directories." | ||||
) | ||||
|
r17524 | # ContentsManager API part 1: methods that must be | ||
|
r15293 | # implemented in subclasses. | ||
|
r13067 | def path_exists(self, path): | ||
"""Does the API-style path (directory) actually exist? | ||||
|
r17523 | |||
|
r17535 | Like os.path.isdir | ||
|
r13070 | Override this method in subclasses. | ||
|
r17523 | |||
|
r13026 | Parameters | ||
---------- | ||||
|
r13067 | path : string | ||
|
r15664 | The path to check | ||
|
r17523 | |||
|
r13026 | Returns | ||
------- | ||||
|
r13067 | exists : bool | ||
Whether the path does indeed exist. | ||||
|
r13026 | """ | ||
|
r13070 | raise NotImplementedError | ||
|
r15097 | |||
def is_hidden(self, path): | ||||
"""Does the API style path correspond to a hidden directory or file? | ||||
|
r17523 | |||
|
r15097 | Parameters | ||
---------- | ||||
path : string | ||||
The path to check. This is an API path (`/` separated, | ||||
|
r17524 | relative to root dir). | ||
|
r17523 | |||
|
r15097 | Returns | ||
------- | ||||
|
r17535 | hidden : bool | ||
|
r15097 | Whether the path is hidden. | ||
|
r17523 | |||
|
r15097 | """ | ||
raise NotImplementedError | ||||
|
r17524 | def file_exists(self, name, path=''): | ||
|
r17535 | """Does a file exist at the given name and path? | ||
Like os.path.isfile | ||||
Override this method in subclasses. | ||||
|
r15291 | |||
Parameters | ||||
---------- | ||||
name : string | ||||
|
r17529 | The name of the file you are checking. | ||
|
r15291 | path : string | ||
|
r17529 | The relative path to the file's directory (with '/' as separator) | ||
|
r15291 | |||
Returns | ||||
------- | ||||
|
r17535 | exists : bool | ||
Whether the file exists. | ||||
|
r15291 | """ | ||
raise NotImplementedError('must be implemented in a subclass') | ||||
|
r17535 | def exists(self, name, path=''): | ||
"""Does a file or directory exist at the given name and path? | ||||
|
r8180 | |||
|
r17535 | Like os.path.exists | ||
|
r8180 | |||
|
r17535 | Parameters | ||
---------- | ||||
name : string | ||||
The name of the file you are checking. | ||||
path : string | ||||
The relative path to the file's directory (with '/' as separator) | ||||
|
r8180 | |||
|
r17535 | Returns | ||
------- | ||||
exists : bool | ||||
Whether the target exists. | ||||
|
r8180 | """ | ||
|
r17535 | return self.file_exists(name, path) or self.path_exists("%s/%s" % (path, name)) | ||
|
r8180 | |||
|
r17524 | def get_model(self, name, path='', content=True): | ||
|
r17529 | """Get the model of a file or directory with or without content.""" | ||
|
r8180 | raise NotImplementedError('must be implemented in a subclass') | ||
|
r17524 | def save(self, model, name, path=''): | ||
|
r17529 | """Save the file or directory and return the model with no content.""" | ||
|
r13046 | raise NotImplementedError('must be implemented in a subclass') | ||
|
r8180 | |||
|
r17524 | def update(self, model, name, path=''): | ||
|
r17535 | """Update the file or directory and return the model with no content. | ||
For use in PATCH requests, to enable renaming a file without | ||||
re-uploading its contents. Only used for renaming at the moment. | ||||
""" | ||||
|
r8180 | raise NotImplementedError('must be implemented in a subclass') | ||
|
r17524 | def delete(self, name, path=''): | ||
|
r17529 | """Delete file or directory by name and path.""" | ||
|
r8180 | raise NotImplementedError('must be implemented in a subclass') | ||
|
r15293 | def create_checkpoint(self, name, path=''): | ||
|
r17529 | """Create a checkpoint of the current state of a file | ||
|
r17523 | |||
|
r15293 | Returns a checkpoint_id for the new checkpoint. | ||
""" | ||||
raise NotImplementedError("must be implemented in a subclass") | ||||
|
r17523 | |||
|
r15293 | def list_checkpoints(self, name, path=''): | ||
|
r17529 | """Return a list of checkpoints for a given file""" | ||
|
r15293 | return [] | ||
|
r17523 | |||
|
r15293 | def restore_checkpoint(self, checkpoint_id, name, path=''): | ||
|
r17529 | """Restore a file from one of its checkpoints""" | ||
|
r15293 | raise NotImplementedError("must be implemented in a subclass") | ||
def delete_checkpoint(self, checkpoint_id, name, path=''): | ||||
|
r17529 | """delete a checkpoint for a file""" | ||
|
r15293 | raise NotImplementedError("must be implemented in a subclass") | ||
|
r17523 | |||
|
r17524 | # ContentsManager API part 2: methods that have useable default | ||
|
r15293 | # implementations, but can be overridden in subclasses. | ||
|
r17535 | def info_string(self): | ||
return "Serving contents" | ||||
|
r16052 | def get_kernel_path(self, name, path='', model=None): | ||
""" Return the path to start kernel in """ | ||||
return path | ||||
|
r17524 | def increment_filename(self, filename, path=''): | ||
"""Increment a filename until it is unique. | ||||
|
r17523 | |||
|
r15293 | Parameters | ||
---------- | ||||
|
r17524 | filename : unicode | ||
The name of a file, including extension | ||||
|
r15293 | path : unicode | ||
|
r17535 | The API path of the target's directory | ||
|
r15293 | |||
Returns | ||||
------- | ||||
name : unicode | ||||
|
r17524 | A filename that is unique, based on the input filename. | ||
|
r15293 | """ | ||
path = path.strip('/') | ||||
|
r17524 | basename, ext = os.path.splitext(filename) | ||
|
r15293 | for i in itertools.count(): | ||
name = u'{basename}{i}{ext}'.format(basename=basename, i=i, | ||||
|
r17524 | ext=ext) | ||
if not self.file_exists(name, path): | ||||
|
r15293 | break | ||
return name | ||||
|
r18249 | def validate_notebook_model(self, model): | ||
"""Add failed-validation message to model""" | ||||
try: | ||||
current.validate(model['content']) | ||||
except current.ValidationError as e: | ||||
model['message'] = 'Notebook Validation failed: {}:\n{}'.format( | ||||
e.message, json.dumps(e.instance, indent=1, default=lambda obj: '<UNKNOWN>'), | ||||
) | ||||
return model | ||||
|
r17527 | def create_file(self, model=None, path='', ext='.ipynb'): | ||
|
r17529 | """Create a new file or directory and return its model with no content.""" | ||
|
r13078 | path = path.strip('/') | ||
|
r13046 | if model is None: | ||
model = {} | ||||
|
r17532 | if 'content' not in model and model.get('type', None) != 'directory': | ||
|
r17527 | if ext == '.ipynb': | ||
metadata = current.new_metadata(name=u'') | ||||
model['content'] = current.new_notebook(metadata=metadata) | ||||
|
r17532 | model['type'] = 'notebook' | ||
model['format'] = 'json' | ||||
|
r17527 | else: | ||
model['content'] = '' | ||||
|
r17532 | model['type'] = 'file' | ||
model['format'] = 'text' | ||||
|
r13074 | if 'name' not in model: | ||
|
r17535 | if model['type'] == 'directory': | ||
untitled = self.untitled_directory | ||||
elif model['type'] == 'notebook': | ||||
untitled = self.untitled_notebook | ||||
elif model['type'] == 'file': | ||||
untitled = self.untitled_file | ||||
else: | ||||
raise HTTPError(400, "Unexpected model type: %r" % model['type']) | ||||
model['name'] = self.increment_filename(untitled + ext, path) | ||||
|
r17523 | |||
|
r13074 | model['path'] = path | ||
|
r17524 | model = self.save(model, model['name'], model['path']) | ||
|
r13046 | return model | ||
|
r8180 | |||
|
r17524 | def copy(self, from_name, to_name=None, path=''): | ||
"""Copy an existing file and return its new model. | ||||
|
r17523 | |||
|
r17532 | If to_name not specified, increment `from_name-Copy#.ext`. | ||
|
r17535 | |||
copy_from can be a full path to a file, | ||||
or just a base name. If a base name, `path` is used. | ||||
|
r13123 | """ | ||
|
r13078 | path = path.strip('/') | ||
|
r17535 | if '/' in from_name: | ||
from_path, from_name = from_name.rsplit('/', 1) | ||||
else: | ||||
from_path = path | ||||
model = self.get_model(from_name, from_path) | ||||
|
r17530 | if model['type'] == 'directory': | ||
raise HTTPError(400, "Can't copy directories") | ||||
|
r13123 | if not to_name: | ||
|
r17524 | base, ext = os.path.splitext(from_name) | ||
copy_name = u'{0}-Copy{1}'.format(base, ext) | ||||
to_name = self.increment_filename(copy_name, path) | ||||
|
r13123 | model['name'] = to_name | ||
|
r17535 | model['path'] = path | ||
|
r17524 | model = self.save(model, to_name, path) | ||
|
r13046 | return model | ||
|
r17523 | |||
|
r15293 | def log_info(self): | ||
self.log.info(self.info_string()) | ||||
|
r15655 | def trust_notebook(self, name, path=''): | ||
|
r15664 | """Explicitly trust a notebook | ||
|
r17523 | |||
|
r15664 | Parameters | ||
---------- | ||||
name : string | ||||
The filename of the notebook | ||||
path : string | ||||
The notebook's directory | ||||
|
r15655 | """ | ||
|
r17525 | model = self.get_model(name, path) | ||
|
r15655 | nb = model['content'] | ||
self.log.warn("Trusting notebook %s/%s", path, name) | ||||
self.notary.mark_cells(nb, True) | ||||
|
r17524 | self.save(model, name, path) | ||
|
r17523 | |||
|
r17535 | def check_and_sign(self, nb, name='', path=''): | ||
|
r15293 | """Check for trusted cells, and sign the notebook. | ||
|
r17523 | |||
|
r15293 | Called as a part of saving notebooks. | ||
|
r17523 | |||
|
r15664 | Parameters | ||
---------- | ||||
nb : dict | ||||
|
r17535 | The notebook object (in nbformat.current format) | ||
|
r15664 | name : string | ||
|
r17535 | The filename of the notebook (for logging) | ||
|
r15664 | path : string | ||
|
r17535 | The notebook's directory (for logging) | ||
|
r10497 | """ | ||
|
r18259 | if nb['nbformat'] != current.nbformat: | ||
return | ||||
|
r15293 | if self.notary.check_cells(nb): | ||
self.notary.sign(nb) | ||||
else: | ||||
self.log.warn("Saving untrusted notebook %s/%s", path, name) | ||||
|
r17523 | |||
|
r17535 | def mark_trusted_cells(self, nb, name='', path=''): | ||
|
r15293 | """Mark cells as trusted if the notebook signature matches. | ||
|
r17523 | |||
|
r15293 | Called as a part of loading notebooks. | ||
|
r17523 | |||
|
r15664 | Parameters | ||
---------- | ||||
nb : dict | ||||
|
r17535 | The notebook object (in nbformat.current format) | ||
|
r15664 | name : string | ||
|
r17535 | The filename of the notebook (for logging) | ||
|
r15664 | path : string | ||
|
r17535 | The notebook's directory (for logging) | ||
|
r15293 | """ | ||
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) | ||||
|
r8181 | |||
|
r15525 | def should_list(self, name): | ||
"""Should this file/directory name be displayed in a listing?""" | ||||
return not any(fnmatch(name, glob) for glob in self.hide_globs) | ||||