notebook-snapshot-extension/snapshot/routes.py

import json

from jupyter_server.base.handlers import APIHandler
from jupyter_server.utils import url_path_join
import tornado

from .store import SnapshotUnchangedError


class HelloRouteHandler(APIHandler):
    # The following decorator should be present on all verb methods (head, get, post,
    # patch, put, delete, options) to ensure only authorized user can request the
    # Jupyter server
    @tornado.web.authenticated
    def get(self):
        self.finish(json.dumps({
            "data": (
                "Hello, world!"
                " This is the '/snapshot/hello' endpoint."
                " Try visiting me in your browser!"
            ),
        }))


class CommitHandler(APIHandler):
    """POST /snapshot/notebook-version/commit

    Persist a new notebook snapshot. Body: ``{path, content, name, description}``.
    """

    @tornado.web.authenticated
    def post(self):
        body = self.get_json_body() or {}
        path = body.get("path")
        content = body.get("content")
        name = body.get("name")
        description = body.get("description", "")
        if not path or content is None or not name:
            self.set_status(400)
            self.finish({"message": "path, content, and name are required"})
            return
        store = self.settings["snapshot_store"]
        try:
            entry = store.commit(path, content, name, description)
        except SnapshotUnchangedError as exc:
            # Not an error — same content as the most recent version.
            # The frontend surfaces this as a warning to the user.
            self.finish({"skipped": True, "reason": "unchanged", "message": str(exc)})
            return
        except ValueError as exc:
            self.set_status(400)
            self.finish({"message": str(exc)})
            return
        except KeyError as exc:
            self.set_status(404)
            self.finish({"message": f"not found: {exc}"})
            return
        self.finish({"entry": entry})


class ListHandler(APIHandler):
    """GET /snapshot/notebook-version/list?path=...

    Return the manifest versions for ``path``, newest first.
    """

    @tornado.web.authenticated
    def get(self):
        path = self.get_query_argument("path", default=None)
        if not path:
            self.set_status(400)
            self.finish({"message": "path is required"})
            return
        versions = self.settings["snapshot_store"].list_versions(path)
        self.finish({"versions": versions})


class ContentHandler(APIHandler):
    """GET /snapshot/notebook-version/content?path=...&id=...

    Return the cleaned notebook body for a given version.
    """

    @tornado.web.authenticated
    def get(self):
        path = self.get_query_argument("path", default=None)
        version_id = self.get_query_argument("id", default=None)
        if not path or not version_id:
            self.set_status(400)
            self.finish({"message": "path and id are required"})
            return
        try:
            notebook = self.settings["snapshot_store"].get_notebook(path, version_id)
        except KeyError:
            self.set_status(404)
            self.finish({"message": "version not found"})
            return
        except ValueError as exc:
            self.set_status(400)
            self.finish({"message": str(exc)})
            return
        self.set_header("Content-Type", "application/json")
        self.finish(json.dumps(notebook))


def setup_route_handlers(web_app):
    host_pattern = ".*$"
    base_url = web_app.settings["base_url"]

    hello_route_pattern = url_path_join(base_url, "snapshot", "hello")
    commit_route_pattern = url_path_join(
        base_url, "snapshot", "notebook-version", "commit"
    )
    list_route_pattern = url_path_join(
        base_url, "snapshot", "notebook-version", "list"
    )
    content_route_pattern = url_path_join(
        base_url, "snapshot", "notebook-version", "content"
    )
    handlers = [
        (hello_route_pattern, HelloRouteHandler),
        (commit_route_pattern, CommitHandler),
        (list_route_pattern, ListHandler),
        (content_route_pattern, ContentHandler),
    ]

    web_app.add_handlers(host_pattern, handlers)