notebook/IPython/html/services/contents/filemanager.py

"""A contents manager that uses the local file system for storage."""

# Copyright (c) IPython Development Team.
# Distributed under the terms of the Modified BSD License.

import io
import os
import glob
import shutil

from tornado import web

from .manager import ContentsManager
from IPython.nbformat import current
from IPython.utils.path import ensure_dir_exists
from IPython.utils.traitlets import Unicode, Bool, TraitError
from IPython.utils.py3compat import getcwd
from IPython.utils import tz
from IPython.html.utils import is_hidden, to_os_path

def sort_key(item):
    """Case-insensitive sorting."""
    return item['name'].lower()


class FileContentsManager(ContentsManager):

    root_dir = Unicode(getcwd(), config=True)

    def _root_dir_changed(self, name, old, new):
        """Do a bit of validation of the root_dir."""
        if not os.path.isabs(new):
            # If we receive a non-absolute path, make it absolute.
            self.root_dir = os.path.abspath(new)
            return
        if not os.path.exists(new) or not os.path.isdir(new):
            raise TraitError("%r is not a directory" % new)

    checkpoint_dir = Unicode('.ipynb_checkpoints', config=True,
        help="""The directory name in which to keep notebook checkpoints

        This is a path relative to the notebook's own directory.

        By default, it is .ipynb_checkpoints
        """
    )

    def _copy(self, src, dest):
        """copy src to dest

        like shutil.copy2, but log errors in copystat
        """
        shutil.copyfile(src, dest)
        try:
            shutil.copystat(src, dest)
        except OSError as e:
            self.log.debug("copystat on %s failed", dest, exc_info=True)

    def get_names(self, path=''):
        """List all filenames in the path (relative to root_dir)."""
        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('*', path))
        names = [ os.path.basename(name) for name in names if os.path.isfile(name)]
        return names

    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 root_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 is_hidden(self, path):
        """Does the API style path correspond to a hidden directory or file?

        Parameters
        ----------
        path : string
            The path to check. This is an API path (`/` separated,
            relative to root_dir).

        Returns
        -------
        exists : bool
            Whether the path is hidden.

        """
        path = path.strip('/')
        os_path = self._get_os_path(path=path)
        return is_hidden(os_path, self.root_dir)

    def _get_os_path(self, name=None, path=''):
        """Given a filename and a URL path, return its file system
        path.

        Parameters
        ----------
        name : string
            A filename
        path : string
            The relative URL path (with '/' as separator) to the named
            file.

        Returns
        -------
        path : string
            API path to be evaluated relative to root_dir.
        """
        if name is not None:
            path = path + '/' + name
        return to_os_path(path, self.root_dir)

    def file_exists(self, name, path=''):
        """Returns a True if the file exists, else returns False.

        Parameters
        ----------
        name : string
            The name of the file you are checking.
        path : string
            The relative path to the file's directory (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'directory does not exist: %r' % os_path)
        elif is_hidden(os_path, self.root_dir):
            self.log.info("Refusing to serve hidden directory, via 404 Error")
            raise web.HTTPError(404, u'directory 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 is_hidden(os_path, self.root_dir)\
                    and self.should_list(name):
                try:
                    model = self.get_dir_model(name, path)
                except IOError:
                    pass
                dirs.append(model)
        dirs = sorted(dirs, key=sort_key)
        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_files(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('/')
        names = self.get_names(path)
        notebooks = [self.get(name, path, content=False)
                        for name in names if self.should_list(name)]
        notebooks = sorted(notebooks, key=sort_key)
        return notebooks

    def get(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.file_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, name, path)
            model['content'] = nb
        return model

    def save(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.file_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(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_name, new_path)

        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))

        model = self.get(new_name, new_path, content=False)
        return model

    def update(self, model, name, path=''):
        """Update the file'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(name, path, new_name, new_path)
        model = self.get(new_name, new_path, content=False)
        return model

    def delete(self, name, path=''):
        """Delete file 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'File 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 file %s", os_path)
        os.unlink(os_path)

    def rename(self, old_name, old_path, new_name, new_path):
        """Rename a file."""
        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)

        # Move the file
        try:
            shutil.move(old_os_path, new_os_path)
        except Exception as e:
            raise web.HTTPError(500, u'Unknown error renaming file: %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)
                shutil.move(old_cp_path, new_cp_path)

    # Checkpoint-related utilities

    def get_checkpoint_path(self, checkpoint_id, name, path=''):
        """find the path to a checkpoint"""
        path = path.strip('/')
        basename, ext = os.path.splitext(name)
        filename = u"{name}-{checkpoint_id}{ext}".format(
            name=basename,
            checkpoint_id=checkpoint_id,
            ext=ext,
        )
        os_path = self._get_os_path(path=path)
        cp_dir = os.path.join(os_path, self.checkpoint_dir)
        ensure_dir_exists(cp_dir)
        cp_path = os.path.join(cp_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 file"""
        path = path.strip('/')
        src_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)
        self._copy(src_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 file

        This contents manager currently only supports one checkpoint per file.
        """
        path = path.strip('/')
        checkpoint_id = "checkpoint"
        os_path = self.get_checkpoint_path(checkpoint_id, name, path)
        if not os.path.exists(os_path):
            return []
        else:
            return [self.get_checkpoint_model(checkpoint_id, name, path)]


    def restore_checkpoint(self, checkpoint_id, name, path=''):
        """restore a file to a checkpointed state"""
        path = path.strip('/')
        self.log.info("restoring %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'checkpoint does not exist: %s-%s' % (name, checkpoint_id)
            )
        # ensure notebook is readable (never restore from an unreadable notebook)
        if cp_path.endswith('.ipynb'):
            with io.open(cp_path, 'r', encoding='utf-8') as f:
                current.read(f, u'json')
        self._copy(cp_path, nb_path)
        self.log.debug("copying %s -> %s", cp_path, nb_path)

    def delete_checkpoint(self, checkpoint_id, name, path=''):
        """delete a file'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'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.root_dir

    def get_kernel_path(self, name, path='', model=None):
        """Return the initial working dir a kernel associated with a given notebook"""
        return os.path.join(self.root_dir, path)