add JCollab Python API

Author: dlqqq

jupyter_server_documents/app.py

@@ -8,6 +8,7 @@
 from .rooms.yroom_manager import YRoomManager
 from .outputs import OutputsManager, outputs_handlers
 from .events import JSD_AWARENESS_EVENT_SCHEMA, JSD_ROOM_EVENT_SCHEMA
+from .jcollab_api import JCollabAPI
 
 class ServerDocsApp(ExtensionApp):
     name = "jupyter_server_documents"
@@ -78,6 +79,14 @@ def get_fileid_manager():
         # Initialize OutputsManager
         self.outputs_manager = self.outputs_manager_class(config=self.config)
         self.settings["outputs_manager"] = self.outputs_manager
+
+        # Serve Jupyter Collaboration API on
+        # `self.settings["jupyter_server_ydoc"]` for compatibility with
+        # extensions depending on Jupyter Collaboration
+        self.settings["jupyter_server_ydoc"] = JCollabAPI(
+            get_fileid_manager=get_fileid_manager,
+            yroom_manager=self.settings["yroom_manager"]
+        )
 
     def _link_jupyter_server_extension(self, server_app):
         """Setup custom config needed by this extension."""

jupyter_server_documents/jcollab_api/__init__.py

@@ -0,0 +1 @@
+from .jcollab_api import JCollabAPI

jupyter_server_documents/jcollab_api/jcollab_api.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+from jupyter_ydoc.ybasedoc import YBaseDoc
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from typing import Callable, Literal
+    from jupyter_server_fileid.manager import BaseFileIdManager
+    from ..rooms import YRoomManager
+
+
+class JCollabAPI:
+    """
+    Provides the Python API provided by `jupyter_collaboration~=4.0` under
+    `self.settings["jupyter_server_ydoc"]`.
+    """
+    fileid_manager: BaseFileIdManager
+    yroom_manager: YRoomManager
+
+    def __init__(self, get_fileid_manager: Callable[[], BaseFileIdManager], yroom_manager: YRoomManager):
+        self._get_fileid_manager = get_fileid_manager
+        self.yroom_manager = yroom_manager
+
+    @property
+    def fileid_manager(self) -> BaseFileIdManager:
+        return self._get_fileid_manager()
+
+    async def get_document(
+        self,
+        *,
+        path: str | None = None,
+        content_type: str | None = None,
+        file_format: Literal["json", "text"] | None = None,
+        room_id: str | None = None,
+        copy: bool = True,
+    ) -> YBaseDoc:
+        """
+        Returns the Jupyter YDoc for a collaborative room.
+
+        You need to provide either a ``room_id`` or the ``path``,
+        the ``content_type`` and the ``file_format``.
+
+        The `copy` argument is ignored by `jupyter_server_documents`.
+        """
+
+        # Raise exception if required arguments are not given
+        if room_id is None and (path is None or content_type is None or file_format is None):
+            raise ValueError(
+                "You need to provide either a ``room_id`` or the ``path``, the ``content_type`` and the ``file_format``."
+            )
+        
+        # Compute room_id if not given
+        if room_id is None:
+            file_id = self.fileid_manager.index(path)
+            room_id = f"{file_format}:{content_type}:{file_id}"
+        
+        # Get or create room using `room_id`
+        room = self.yroom_manager.get_room(room_id)
+        if not room:
+            raise ValueError(
+                f"Could not get room using room ID '{room_id}'."
+            )
+        
+        # Return the Jupyter YDoc once ready
+        return await room.get_jupyter_ydoc()