Added set_attribute method and added support for subclassing of Dispo…

…sableConnection with CommandConnection as the first. Commands can be enabled disabled and 'toggled' via configure.

Added missingAsNull for get_attribute.

ipylab/_compat/enum.py

@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+import sys
+
+if sys.version_info < (3, 11):
+    from backports.strenum import StrEnum
+else:
+    from enum import StrEnum
+
+__all__ = ["StrEnum"]
+
+
+def __dir__() -> list[str]:
+    return __all__

ipylab/_compat/typing.py

@@ -0,0 +1,17 @@
+from __future__ import annotations
+
+import sys
+from typing import TYPE_CHECKING, Any, Optional, TypeAlias, TypedDict
+
+if sys.version_info < (3, 11):
+    from typing import Optional
+
+    from typing_extensions import Self, Unpack
+else:
+    from typing import Optional, Self, Unpack
+
+__all__ = ["Optional", "TYPE_CHECKING", "Any", "TypeAlias", "TypedDict", "Self", "Unpack"]
+
+
+def __dir__() -> list[str]:
+    return __all__

ipylab/asyncwidget.py

@@ -4,7 +4,6 @@
 
 import asyncio
 import inspect
-import sys
 import traceback
 import uuid
 from typing import TYPE_CHECKING, Any
@@ -13,18 +12,14 @@
 from traitlets import Container, Dict, Instance, Set, Unicode
 
 import ipylab._frontend as _fe
+from ipylab._compat.enum import StrEnum
 from ipylab.hasapp import HasApp
 from ipylab.hookspecs import pm
 
-if sys.version_info >= (3, 11):
-    from enum import StrEnum
-else:
-    from backports.strenum import StrEnum
-
 if TYPE_CHECKING:
     import logging
     from collections.abc import Coroutine, Iterable
-    from typing import ClassVar
+    from typing import ClassVar, Literal
 
 
 __all__ = ["AsyncWidgetBase", "WidgetBase", "register", "pack", "Widget"]
@@ -384,9 +379,43 @@ def execute_method(
             toLuminoWidget=toLuminoWidget,
         )
 
-    def get_attribute(self, path: str, *, transform: TransformType = TransformMode.raw):
+    def get_attribute(
+        self,
+        path: str,
+        *,
+        transform: TransformType = TransformMode.raw,
+        ifMissing: Literal["raise", "null"] = "raise",
+    ):
         """A serialized version of the attribute relative to this object."""
-        return self.execute_method("getAttribute", path, transform=transform)
+        return self.execute_method("getAttribute", path, ifMissing, transform=transform)
+
+    def set_attribute(
+        self,
+        path: str,
+        value,
+        valueTransform: TransformType = TransformMode.raw,
+        *,
+        valueToLuminoWidget=False,
+    ):
+        """Set the attribute at the path in the frontend.
+        path: str
+            "the.path.to.the.attribute" to be set.
+        value: jsonable
+            The value to set, or instructions for the transform to do in the frontend.
+        valueTransform: TransformType
+            valueTransform is applied to the value prior to setting the attribute.
+        valueToLuminoWidget: bool
+            Whether the value should be converted to a Lumino widget. The value transform
+            can be left as raw unless further adanced transformation is required.
+        """
+        return self.execute_method(
+            "setAttribute",
+            path,
+            pack(value),
+            valueTransform,
+            toLuminoWidget=["args[1]"] if valueToLuminoWidget else [],
+            transform=TransformMode.done,
+        )
 
     def list_methods(self, path: str = "", *, depth=2, skip_hidden=True):
         """Get a list of methods belonging to the object 'path' of the Frontend instance.

ipylab/commands.py

@@ -3,30 +3,73 @@
 from __future__ import annotations
 
 import inspect
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING
 
-from traitlets import Dict, Tuple, Unicode
+from traitlets import Callable as CallableTrait
+from traitlets import Tuple, Unicode
 
+from ipylab._compat.typing import Any, Optional, TypedDict, Unpack
 from ipylab.asyncwidget import AsyncWidgetBase, TransformMode, pack, register
-from ipylab.hookspecs import pm
+from ipylab.disposable_connection import DisposableConnection
 from ipylab.jupyterfrontend_subsection import FrontEndSubsection
 
 if TYPE_CHECKING:
-    from collections.abc import Callable
+    from asyncio import Task
+    from collections.abc import Callable, Coroutine
 
     from ipylab.widgets import Icon
 
 
+class CommandOptions(TypedDict):
+    enabled: Optional[bool]
+    visible: Optional[bool]
+    toggled: Optional[bool]
+
+
+class CommandConnection(DisposableConnection):
+    """A Disposable Ipylab command registered in the command pallet."""
+
+    ID_PREFIX = "ipylab_command"
+    python_command = CallableTrait(allow_none=False)
+
+    def configure(self, *, emit=True, **kwgs: Unpack[CommandOptions]) -> Task[CommandOptions]:
+        async def configure_():
+            config = await self.get_config()
+            for k, v in kwgs.items():
+                if v is not None:
+                    config[k] = v
+            await self.set_attribute("config", config)
+            if emit:
+                await self.app.commands.execute_method("commandChanged.emit", {"id": self.id})
+            return config
+
+        return self.to_task(configure_())
+
+    def get_config(self) -> Task[CommandOptions]:
+        async def get_config_():
+            config = await self.get_attribute("config", ifMissing="null")
+            return config or {}
+
+        return self.to_task(get_config_())
+
+
 @register
 class CommandPalette(AsyncWidgetBase):
     _model_name = Unicode("CommandPaletteModel").tag(sync=True)
     items = Tuple(read_only=True).tag(sync=True)
 
-    def add_item(self, command_id: str, category: str, *, rank=None, args: dict | None = None):
-        return self.schedule_operation(operation="addItem", id=command_id, category=category, rank=rank, args=args)
+    def add_item(self, command_id: str | CommandConnection, category: str, *, rank=None, args: dict | None = None):
+        return self.schedule_operation(
+            operation="addItem",
+            id=str(command_id),
+            category=category,
+            rank=rank,
+            args=args,
+            transform=TransformMode.connection,
+        )
 
-    def remove_item(self, command_id: str, category):
-        return self.schedule_operation(operation="removeItem", id=command_id, category=category)
+    def remove_item(self, command_id: str | CommandConnection, category):
+        return self.schedule_operation(operation="removeItem", id=str(command_id), category=category)
 
 
 @register
@@ -40,67 +83,71 @@ class CommandRegistry(FrontEndSubsection):
     SINGLETON = True
     SUB_PATH_BASE = "app.commands"
     all_commands = Tuple(read_only=True).tag(sync=True)
-    _execute_callbacks: Dict[str, Callable[[], None]] = Dict()
 
     async def _do_operation_for_frontend(self, operation: str, payload: dict, buffers: list) -> Any:
         match operation:
             case "execute":
-                command_id: str = payload.get("id")  # type:ignore
-                cmd = self._get_command(command_id)
+                conn = self.get_existing_command_connection(payload["id"])
+                if not conn:
+                    msg = f"Command not found with id='{payload['id']}'!"
+                    raise RuntimeError(msg)
+                cmd = conn.python_command
                 kwgs = payload.get("kwgs") or {} | {"buffers": buffers}
                 for k in set(kwgs).difference(inspect.signature(cmd).parameters.keys()):
                     kwgs.pop(k)
                 result = cmd(**kwgs)
                 if inspect.isawaitable(result):
                     return await result
                 return result
-            case _:
-                pm.hook.unhandled_frontend_operation_message(obj=self, operation=operation)
-        raise NotImplementedError
-
-    def _get_command(self, command_id: str) -> Callable:
-        "Get a registered Python command"
-        if command_id not in self._execute_callbacks:
-            msg = f"{command_id} is not a registered command!"
-            raise KeyError(msg)
-        return self._execute_callbacks[command_id]
+        return await super()._do_operation_for_frontend(operation, payload, buffers)
 
     def add_command(
         self,
-        command_id: str,
-        execute: Callable,
+        name: str,
+        execute: Callable[..., Coroutine | Any],
         *,
         caption="",
         label="",
         icon_class="",
         icon: Icon | None = None,
-    ):
-        """
-        toLuminoWidget : bool
-            If the result should be transformed into a luminoWidget in the frontend.
+        command_result_transform: TransformMode = TransformMode.done,
+        **kwgs,
+    ) -> Task[CommandConnection]:
+        """Add a python command that can be executed by Jupyterlab.
+
+        name: str
+            The suffix for the 'id'.
+        execute:
+
+        kwgs:
+            Additional ICommandOptions can be passed as kwgs
+
+        ref: https://lumino.readthedocs.io/en/latest/api/interfaces/commands.CommandRegistry.ICommandOptions.html
         """
-        # TODO: support other parameters (isEnabled, isVisible...)
-        self._execute_callbacks = self._execute_callbacks | {command_id: execute}
-        return self.schedule_operation(
+        task = self.schedule_operation(
             "add_command",
-            id=command_id,
+            id=CommandConnection.to_id(name),
             caption=caption,
             label=label,
             iconClass=icon_class,
-            icon=pack(icon),
             transform=TransformMode.connection,
+            icon=pack(icon),
+            command_result_transform=command_result_transform,
+            **kwgs,
         )
 
-    def remove_command(self, command_id: str):
-        # TODO: check whether to keep this method, or return disposables like in lab
-        if command_id not in self._execute_callbacks:
-            msg = f"{command_id=} is not a registered command!"
-            raise ValueError(msg)
+        async def add_command_():
+            conn: CommandConnection = await task
+            conn.set_trait("python_command", execute)
+            return conn
 
-        task = self.schedule_operation("remove_command", command_id=command_id, transform=TransformMode.done)
+        return self.to_task(add_command_())
 
-        async def remove_command_():
-            await task
-            self._execute_callbacks.pop(command_id, None)
+    def remove_command(self, name_or_id: str):
+        comm = self.get_existing_command_connection(name_or_id)
+        if comm:
+            comm.dispose()
 
-        return self.to_task(remove_command_())
+    def get_existing_command_connection(self, name_or_id: str) -> CommandConnection | None:
+        "Will return a CommandConnection if it was added in this kernel."
+        return CommandConnection.get_existing_connection(name_or_id)

ipylab/disposable_connection.py

@@ -5,17 +5,19 @@
 
 import asyncio
 import contextlib
-from typing import ClassVar
+from typing import Generic, TypeVar
 
 from ipywidgets import register
 from traitlets import Unicode
 
 from ipylab.asyncwidget import AsyncWidgetBase
 from ipylab.jupyterfrontend_subsection import FrontEndSubsection
 
+T = TypeVar("T", bound="DisposableConnection")
+
 
 @register
-class DisposableConnection(FrontEndSubsection, AsyncWidgetBase):
+class DisposableConnection(FrontEndSubsection, AsyncWidgetBase, Generic[T]):
     """A connection to a disposable object in the Frontend.
 
     The dispose method is directly accesssable.
@@ -26,18 +28,43 @@ class DisposableConnection(FrontEndSubsection, AsyncWidgetBase):
 
     see: https://lumino.readthedocs.io/en/latest/api/modules/disposable.html
 
+    Subclasses that are inherited with and ID_PREFIX
     """
 
     SUB_PATH_BASE = "obj"
-    _connections: ClassVar[dict[str, DisposableConnection]] = {}
+    ID_PREFIX = ""
+    _CLASS_DEFINITIONS: dict[str, type[T]] = {}  # noqa RUF012
+    _connections: dict[str, T] = {}  # noqa RUF012
     _model_name = Unicode("DisposableConnectionModel").tag(sync=True)
     id = Unicode(read_only=True).tag(sync=True)
 
-    def __new__(cls, *, id: str, **kwgs):  # noqa: A002
+    def __init_subclass__(cls, **kwargs) -> None:
+        if cls.ID_PREFIX:
+            cls._CLASS_DEFINITIONS[cls.ID_PREFIX] = cls  # type: ignore
+        super().__init_subclass__(**kwargs)
+
+    def __new__(cls, *, id: str, **kwgs):  # noqa A002
         if id not in cls._connections:
-            cls._connections[id] = super().__new__(cls, **kwgs)
+            if cls.ID_PREFIX and not id.startswith(cls.ID_PREFIX):
+                msg = f"Expected prefix '{cls.ID_PREFIX}' not found for {id=}"
+                raise ValueError(msg)
+            # Check if a subclass is registered with 'ID_PREFIX'
+            cls_ = cls._CLASS_DEFINITIONS.get(id.split(":")[0], cls) if ":" in id else cls
+            cls._connections[id] = super().__new__(cls_, **kwgs)  # type: ignore
         return cls._connections[id]
 
+    def __str__(self):
+        return self.id
+
+    @classmethod
+    def to_id(cls, name_or_id: str | T) -> str:
+        """Generate an id for the given name."""
+        if isinstance(name_or_id, DisposableConnection):
+            return name_or_id.id
+        if not cls.ID_PREFIX:
+            return name_or_id
+        return f"{cls.ID_PREFIX}:{name_or_id.removeprefix(cls.ID_PREFIX).strip(':')}"
+
     def __init__(self, *, id: str, model_id=None, **kwgs):  # noqa: A002
         if self._async_widget_base_init_complete:
             return
@@ -57,3 +84,8 @@ async def dispose_():
                     await self.execute_method("dispose")
 
         return self.to_task(dispose_())
+
+    @classmethod
+    def get_existing_connection(cls, name_or_id: str | T):
+        "Get an existing connection"
+        return cls._connections.get(cls.to_id(name_or_id))

ipylab/jupyterfrontend.py

@@ -76,7 +76,6 @@ def _init_python_backend(self):
         "Run by the Ipylab python backend."
         # This is called in a separate kernel started by the JavaScript frontend
         # the first time the ipylab plugin is activated.
-        from ipylab.hookspecs import pm
 
         try:
             count = pm.load_setuptools_entrypoints("ipylab_backend")
@@ -89,9 +88,7 @@ async def _do_operation_for_frontend(self, operation: str, payload: dict, buffer
         match operation:
             case "execEval":
                 return await self._exec_eval(payload, buffers)
-            case _:
-                pm.hook.unhandled_frontend_operation_message(obj=self, operation=operation)
-        raise NotImplementedError
+        return await super()._do_operation_for_frontend(operation, payload, buffers)
 
     def shutdown_kernel(self, kernelId: str | None = None):
         """Shutdown the kernel"""