Add first implementation for list and get variables (#12)

* Fix linter issue

* Add first iteration for list and get variables

* Please pre-commit

* Test forcing mimetype

* Automatic application of license header

---------

Co-authored-by: Frédéric Collonval <[email protected]>
Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

.github/workflows/fix-license-header.yml

@@ -6,7 +6,7 @@ on:
 concurrency:
   group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
   cancel-in-progress: true
-    
+
 jobs:
   header-license-fix:
     runs-on: ubuntu-latest

.licenserc.yaml

@@ -14,4 +14,4 @@ header:
     - '**/.*'
     - 'LICENSE'
 
-  comment: on-failure
+  comment: on-failure

README.md

@@ -17,7 +17,7 @@ Jupyter Kernel Client allows you to connect via WebSocket and HTTP to Jupyter Ke
 
 To install the library, run the following command.
 
-```bash
+```sh
 pip install jupyter_kernel_client
 ```
 
@@ -31,13 +31,13 @@ pip install jupyter-server ipykernel
 
 ### Kernel Client
 
-2. Start a Jupyter Server.
+1. Start a Jupyter Server.
 
 ```sh
 jupyter server --port 8888 --IdentityProvider.token MY_TOKEN
 ```
 
-3. Launch `python` in a terminal and execute the following snippet (update the server_url and token).
+2. Launch `python` in a terminal and execute the following snippet (update the server_url and token).
 
 ```py
 import os
@@ -77,13 +77,16 @@ pip install jupyter-kernel-client[konsole]
 2. Start a Jupyter Server.
 
 ```sh
-jupyter server --port 8888 --token MY_TOKEN
+jupyter server --port 8888 --IdentityProvider.token MY_TOKEN
 ```
 
 3. Start the konsole and execute code.
 
-```bash
-$ jupyter konsole --url http://localhost:8888 --IdentityProvider.token MY_TOKEN
+```sh
+jupyter konsole --url http://localhost:8888 --token MY_TOKEN
+```
+
+```sh
 [KonsoleApp] KernelHttpManager created a new kernel: ...
 Jupyter Kernel console 0.2.0
 
@@ -94,22 +97,22 @@ IPython 8.30.0 -- An enhanced Interactive Python. Type '?' for help.
 In [1]: print("hello")
 hello
 
-In [2]:                                                                                                  
+In [2]:
 ```
 
 ## Uninstall
 
 To remove the library, execute:
 
-```bash
+```sh
 pip uninstall jupyter_kernel_client
 ```
 
 ## Contributing
 
 ### Development install
 
-```bash
+```sh
 # Clone the repo to your local environment
 # Change directory to the jupyter_kernel_client directory
 # Install package in development mode - will automatically enable
@@ -121,19 +124,19 @@ pip install -e ".[konsole,test,lint,typing]"
 
 Install dependencies:
 
-```bash
+```sh
 pip install -e ".[test]"
 ```
 
 To run the python tests, use:
 
-```bash
+```sh
 pytest
 ```
 
 ### Development uninstall
 
-```bash
+```sh
 pip uninstall jupyter_kernel_client
 ```
 

jupyter_kernel_client/__init__.py

@@ -7,12 +7,14 @@
 from .client import KernelClient
 from .konsoleapp import KonsoleApp
 from .manager import KernelHttpManager
+from .snippets import SNIPPETS_REGISTRY, LanguageSnippets
 from .wsclient import KernelWebSocketClient
 
-
 __all__ = [
+    "SNIPPETS_REGISTRY",
     "KernelClient",
     "KernelHttpManager",
     "KernelWebSocketClient",
     "KonsoleApp",
+    "LanguageSnippets",
 ]

jupyter_kernel_client/__version__.py

@@ -4,5 +4,4 @@
 
 """Jupyter Kernel Client through websocket."""
 
-
 __version__ = "0.3.1"

jupyter_kernel_client/client.py

@@ -14,6 +14,7 @@
 
 from .constants import REQUEST_TIMEOUT
 from .manager import KernelHttpManager
+from .snippets import SNIPPETS_REGISTRY
 from .utils import UTC
 
 logger = logging.getLogger("jupyter_kernel_client")
@@ -149,7 +150,7 @@ class KernelClient(LoggingConfigurable):
             Client user name; default to environment variable USER
         kernel_id: str | None
             ID of the kernel to connect to
-    """
+    """  # noqa E501
 
     kernel_manager_class = Type(
         default_value=KernelHttpManager,
@@ -187,6 +188,18 @@ def id(self) -> str | None:
         """Kernel ID"""
         return self._manager.kernel["id"] if self._manager.kernel else None
 
+    @property
+    def kernel_info(self) -> dict[str, t.Any] | None:
+        """Kernel information.
+
+        This is the dictionary returned by the kernel for a kernel_info_request.
+
+        Returns:
+            The kernel information
+        """
+        if self._manager.kernel:
+            return self._manager.client.kernel_info_interactive(timeout=REQUEST_TIMEOUT)
+
     @property
     def last_activity(self) -> datetime.datetime | None:
         """Kernel process last activity.
@@ -337,7 +350,7 @@ def restart(self, timeout: float = REQUEST_TIMEOUT) -> None:
         """Restarts a kernel."""
         return self._manager.restart_kernel(timeout=timeout)
 
-    def __enter__(self) -> "KernelClient":
+    def __enter__(self) -> KernelClient:
         self.start()
         return self
 
@@ -383,3 +396,91 @@ def stop(
             shutdown = self._own_kernel if shutdown_kernel is None else shutdown_kernel
             if shutdown:
                 self._manager.shutdown_kernel(now=shutdown_now, timeout=timeout)
+
+    #
+    # Variables related methods
+    #
+    def get_variable(self, name: str, mimetype: str | None = None) -> dict[str, t.Any]:
+        """Get a kernel variable.
+
+        Args:
+            name: Variable name
+            mimetype: optional, type of variable value serialization; default ``None``,
+                i.e. returns all known serialization.
+
+        Returns:
+            A dictionary for which keys are mimetype and values the variable value
+            serialized in that mimetype.
+            Even if a mimetype is specified, the dictionary may not contain it if
+            the kernel introspection failed to get the variable in the specified format.
+        Raises:
+            ValueError: If the kernel programming language is not supported
+            RuntimeError: If the kernel introspection failed
+        """
+        kernel_language = (self.kernel_info or {}).get("language_info", {}).get("name")
+        if kernel_language not in SNIPPETS_REGISTRY.available_languages:
+            raise ValueError(f"""Code snippet for language {kernel_language} are not available.
+You can set them yourself using:
+
+    from jupyter_kernel_client import SNIPPETS_REGISTRY, LanguageSnippets
+    SNIPPETS_REGISTRY.register("my-language", LanguageSnippets(list_variables="", get_variable=""))
+""")
+
+        snippet = SNIPPETS_REGISTRY.get_get_variable(kernel_language)
+        results = self.execute(snippet.format(name=name, mimetype=mimetype), silent=True)
+
+        self.log.debug("Kernel variables: %s", results)
+
+        if results["status"] == "ok" and results["outputs"]:
+            if mimetype is None:
+                return results["outputs"][0]["data"]
+            else:
+                has_mimetype = mimetype in results["outputs"][0]["data"]
+                return {mimetype: results["outputs"][0]["data"][mimetype]} if has_mimetype else {}
+        else:
+            raise RuntimeError(f"Failed to get variable {name} with type {mimetype}.")
+
+    def list_variables(self) -> list[dict[str, t.Any]]:
+        """List the kernel global variables.
+
+        A variable is defined by a dictionary with the schema:
+        {
+            "type": "object",
+            "properties": {
+                "name": {"title": "Variable name", "type": "string"},
+                "type": {"title": "Variable type", "type": "string"},
+                "size": {"title": "Variable size in bytes.", "type": "number"}
+            },
+            "required": ["name", "type"]
+        }
+
+        Returns:
+            The list of global variables.
+        Raises:
+            ValueError: If the kernel programming language is not supported
+            RuntimeError: If the kernel introspection failed
+        """
+        kernel_language = (self.kernel_info or {}).get("language_info", {}).get("name")
+        if kernel_language not in SNIPPETS_REGISTRY.available_languages:
+            raise ValueError(f"""Code snippet for language {kernel_language} are not available.
+You can set them yourself using:
+
+    from jupyter_kernel_client import SNIPPETS_REGISTRY, LanguageSnippets
+    SNIPPETS_REGISTRY.register("my-language", LanguageSnippets(list_variables="", get_variable=""))
+""")
+
+        snippet = SNIPPETS_REGISTRY.get_list_variables(kernel_language)
+        results = self.execute(snippet, silent=True)
+
+        self.log.debug("Kernel variables: %s", results)
+
+        if (
+            results["status"] == "ok"
+            and results["outputs"]
+            and "application/json" in results["outputs"][0]["data"]
+        ):
+            return sorted(
+                results["outputs"][0]["data"]["application/json"], key=lambda v: v["name"]
+            )
+        else:
+            raise RuntimeError("Failed to list variables.")

jupyter_kernel_client/konsoleapp.py

@@ -16,7 +16,6 @@
 from .manager import KernelHttpManager
 from .shell import WSTerminalInteractiveShell
 
-
 # -----------------------------------------------------------------------------
 # Globals
 # -----------------------------------------------------------------------------
@@ -27,7 +26,7 @@
 
 # Start a console connected to a distant Jupyter Server with a new python kernel.
 jupyter konsole --url https://my.jupyter-server.xzy --token <server_token>
-"""
+"""  # noqa E501
 
 # -----------------------------------------------------------------------------
 # Flags and Aliases
@@ -102,7 +101,7 @@ class KonsoleApp(JupyterApp):
     """
     examples = _examples
 
-    classes = [WSTerminalInteractiveShell]
+    classes = [WSTerminalInteractiveShell]  # noqa RUF012
     flags = Dict(flags)
     aliases = Dict(aliases)
 

jupyter_kernel_client/shell.py

@@ -34,7 +34,7 @@ async def handle_external_iopub(self, loop=None):
                 await asyncio.sleep(0.5)
 
         def show_banner(self):
-            print(
+            print(  # noqa T201
                 self.banner.format(
                     version=__version__, kernel_banner=self.kernel_info.get("banner", "")
                 ),
@@ -57,7 +57,7 @@ def __init__(self):
             self._executing = False
 
         def show_banner(self):
-            return "You must install `jupyter_console` to use the console:\n\n\tpip install jupyter-console\n"
+            return "You must install `jupyter_console` to use the console:\n\n\tpip install jupyter-console\n"  # noqa E501
 
         def mainloop(self) -> None:
             raise ModuleNotFoundError("jupyter_console")