ContentsManager strawman

examples/contents.ipynb

@@ -0,0 +1,186 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "id": "b0fe59fc-2f42-4ee5-b1cf-d3a73ccd80b2",
+   "metadata": {},
+   "source": [
+    "# Contents"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "531ea466-29a1-43b9-87bd-99432eaf9239",
+   "metadata": {},
+   "source": [
+    "### ContentsManager"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "3854222d-d955-416f-8c19-acda05c940bc",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import sys\n",
+    "import uuid\n",
+    "from IPython.display import JSON\n",
+    "\n",
+    "path = \"examples/contents.ipynb\"\n",
+    "if \"pyodide\" in sys.modules:\n",
+    "    %pip install ipylab\n",
+    "    path = \"contents.ipynb\""
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "1bf0d356-1e8b-4486-bf87-6086919c629e",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from ipylab import JupyterFrontEnd"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "db015374-352c-442b-af4e-81bc8d5e6db8",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "app = JupyterFrontEnd()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "aff5288d-2cf6-4304-ba94-966480174534",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "app.contents"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "d3dfcf18-c84d-4d5c-b4c9-1d1d37ad15ca",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "this_notebook = app.contents.get(path, content=True)\n",
+    "this_notebook"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "23cdb103-2b14-4528-b13f-567c0afa38bc",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "JSON(this_notebook.content)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "ec496589-101e-48f8-ac60-0c84238cd62a",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "new_content = dict(this_notebook.content.items())\n",
+    "new_content.update(\n",
+    "    cells=[\n",
+    "        {\n",
+    "            'cell_type': 'code',\n",
+    "            'execution_count': None,\n",
+    "            'id': str(__import__(\"uuid\").uuid4()),\n",
+    "            'metadata': {},\n",
+    "            'outputs': [],\n",
+    "            'source': 'print(\"HELLO WORLD\")'\n",
+    "        },\n",
+    "        *new_content[\"cells\"]\n",
+    "    ]\n",
+    ")\n",
+    "JSON(new_content)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "61c0a006-d40d-43fb-b67e-c2308da92344",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "this_notebook.content = new_content"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "ad137fa0-a5f8-48ac-ada0-0dbda9cb803e",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "new_content = dict(this_notebook.content.items())\n",
+    "new_content.update(\n",
+    "    cells=[\n",
+    "        {\n",
+    "            'cell_type': 'code',\n",
+    "            'execution_count': None,\n",
+    "            'id': str(__import__(\"uuid\").uuid4()),\n",
+    "            'metadata': {},\n",
+    "            'outputs': [],\n",
+    "            'source': 'print(\"HELLO WORLD\")'\n",
+    "        },\n",
+    "        *new_content[\"cells\"]\n",
+    "    ]\n",
+    ")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "65053177-480d-4d94-a08f-93d3f25e13f7",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "this_notebook.content = new_content"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "4f632caa-6625-46af-bb95-2fcd812e2919",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "this_notebook.content"
+   ]
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Python 3 (ipykernel)",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.11.6"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 5
+}

ipylab/contents.py

@@ -0,0 +1,113 @@
+#!/usr/bin/env python
+# coding: utf-8
+
+# Copyright (c) ipylab contributors.
+# Distributed under the terms of the Modified BSD License.
+
+import typing as t
+from uuid import uuid4
+from ipywidgets import Widget, register
+from traitlets import HasTraits, Unicode, Any, Instance, observe, Int, Bool
+from ._frontend import module_name, module_version
+
+
+@register
+class ContentsManager(Widget):
+    _model_name = Unicode("ContentsManagerModel").tag(sync=True)
+    _model_module = Unicode(module_name).tag(sync=True)
+    _model_module_version = Unicode(module_version).tag(sync=True)
+
+    _requests: t.Dict[str, t.Callable]
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._requests = {}
+        self.on_msg(self._on_frontend_msg)
+
+    def _on_frontend_msg(self, _, content, buffers):
+        _id = content.get("_id")
+        callback = self._requests.pop(_id, None)
+        if callback:
+            callback(content)
+
+    def get(self, path: str, content: t.Optional[bool] = None) -> "ContentsModel":
+        _id = str(uuid4())
+        self.send(
+            {
+                "_id": _id,
+                "func": "get",
+                "payload": {"path": path, "options": {"content": content}},
+            }
+        )
+
+        model = ContentsModel(path=path, _contents_manager=self)
+
+        self._requests[_id] = model._on_get
+
+        return model
+
+    def save(self, model: "ContentsModel"):
+        _id = str(uuid4())
+        self.send(
+            {
+                "_id": _id,
+                "func": "save",
+                "payload": {
+                    "path": model.path,
+                    "options": {
+                        k: getattr(model, k)
+                        for k in model.class_trait_names()
+                        if not (
+                            k.startswith("_") or k == "error" or k in Widget._traits
+                        )
+                    },
+                },
+            }
+        )
+
+        self._requests[_id] = model._on_get
+
+        return model
+
+
+@register
+class ContentsModel(Widget):
+    _model_name = Unicode("ContentsModelModel").tag(sync=True)
+    _model_module = Unicode(module_name).tag(sync=True)
+    _model_module_version = Unicode(module_version).tag(sync=True)
+    _contents_manager = Instance(ContentsManager)
+    _syncing: t.Optional[bool]
+
+    error = Unicode(allow_none=True)
+
+    name = Unicode(allow_none=True)
+    path = Unicode(allow_none=True).tag(sync=True)
+    last_modified = Unicode(allow_none=True)
+    created = Unicode(allow_none=True)
+    format = Unicode(allow_none=True)
+    mimetype = Unicode(allow_none=True)
+    size = Int(allow_none=True)
+    writeable = Bool(allow_none=True)
+    type = Unicode(allow_none=True)
+    content = Any(allow_none=True)
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self.observe(self._on_content, "content")
+
+    def _on_get(self, msg):
+        if "error" in msg:
+            self.error = msg["error"]
+            return
+        func = msg.get("func")
+        self._syncing = True
+        for key, value in msg.get("model", {}).items():
+            if func == "save" and key == "content":
+                continue
+            setattr(self, key, value)
+        self._syncing = False
+
+    def _on_content(self, change) -> None:
+        if self._syncing:
+            return
+        self._contents_manager.save(self)

ipylab/jupyterfrontend.py

@@ -13,6 +13,7 @@
 from .commands import CommandRegistry
 from .shell import Shell
 from .sessions import SessionManager
+from .contents import ContentsManager
 
 
 @register
@@ -25,13 +26,15 @@ class JupyterFrontEnd(Widget):
     shell = Instance(Shell).tag(sync=True, **widget_serialization)
     commands = Instance(CommandRegistry).tag(sync=True, **widget_serialization)
     sessions = Instance(SessionManager).tag(sync=True, **widget_serialization)
+    contents = Instance(ContentsManager).tag(sync=True, **widget_serialization)
 
     def __init__(self, *args, **kwargs):
         super().__init__(
             *args,
             shell=Shell(),
             commands=CommandRegistry(),
             sessions=SessionManager(),
+            contents=ContentsManager(),
             **kwargs,
         )
         self._ready_event = asyncio.Event()

src/plugin.ts

@@ -37,6 +37,8 @@ const extension: JupyterFrontEndPlugin<void> = {
 
         // add globals
         widgetExports.JupyterFrontEndModel.app = app;
+        widgetExports.ContentsManagerModel.contentsManager =
+          app.serviceManager.contents;
         widgetExports.ShellModel.shell = app.shell;
         widgetExports.ShellModel.labShell = labShell;
         widgetExports.CommandRegistryModel.commands = app.commands;

src/widget.ts

@@ -5,6 +5,7 @@ import { CommandRegistryModel } from './widgets/commands';
 import { CommandPaletteModel } from './widgets/palette';
 import { SessionManagerModel } from './widgets/sessions';
 import { JupyterFrontEndModel } from './widgets/frontend';
+import { ContentsManagerModel, ContentsModelModel } from './widgets/contents';
 import { PanelModel } from './widgets/panel';
 import { ShellModel } from './widgets/shell';
 import { SplitPanelModel, SplitPanelView } from './widgets/split_panel';
@@ -14,6 +15,8 @@ import { IconView, IconModel } from './widgets/icon';
 export {
   CommandRegistryModel,
   CommandPaletteModel,
+  ContentsManagerModel,
+  ContentsModelModel,
   JupyterFrontEndModel,
   PanelModel,
   ShellModel,