Shift contextvars into utils notably, removing kernel.job.

Add Kernel.concurrency_mode replacing  get_handler_and_run_mode with get_run_mode and get_handler.
Add more methods to utils.

Author: fleming79

docs/notebooks/simple_example.ipynb

@@ -40,9 +40,8 @@
    "source": [
     "import anyio\n",
     "import ipywidgets as ipw\n",
-    "import utils\n",
     "\n",
-    "from async_kernel import Caller\n",
+    "from async_kernel import Caller, utils\n",
     "\n",
     "\n",
     "async def demo():\n",
@@ -53,7 +52,7 @@
     "    for i in range(1, 4):\n",
     "        b.description = f\"Continue {i}\"\n",
     "        event = anyio.Event()\n",
-    "        b.on_click(lambda _: caller.call_soon(event.set))  # noqa: B023\n",
+    "        b.on_click(lambda _: caller.call_soon(event.set))  # noqa: B023  # pyright: ignore[reportUnknownLambdaType]\n",
     "        print(f\"Waiting {i}\", end=\"\\r\")\n",
     "        await event.wait()\n",
     "    b.close()\n",
@@ -156,7 +155,9 @@
     "slideshow": {
      "slide_type": ""
     },
-    "tags": []
+    "tags": [
+     "suppress-error"
+    ]
    },
    "outputs": [],
    "source": [

src/async_kernel/asyncshell.py

@@ -7,7 +7,6 @@
 import json
 import pathlib
 import sys
-from contextvars import ContextVar
 from typing import TYPE_CHECKING, Any, ClassVar, Literal
 
 import anyio
@@ -22,6 +21,7 @@
 from typing_extensions import override
 
 import async_kernel
+from async_kernel import utils
 from async_kernel.caller import Caller
 from async_kernel.compiler import XCachingCompiler
 from async_kernel.typing import Content, Tags
@@ -97,7 +97,7 @@ def publish(  # pyright: ignore[reportIncompatibleMethodOverride]
 
         [Reference](https://jupyter-client.readthedocs.io/en/stable/messaging.html#update-display-data)
         """
-        async_kernel.Kernel().iopub_send(
+        utils.get_kernel().iopub_send(
             msg_or_type="update_display_data" if update else "display_data",
             content={"data": data, "metadata": metadata or {}, "transient": transient or {}} | kwargs,
             ident=self.topic,
@@ -112,7 +112,7 @@ def clear_output(self, wait: bool = False) -> None:
                 instead waiting for the next display before clearing.
                 This reduces bounce during repeated clear & display loops.
         """
-        async_kernel.Kernel().iopub_send(msg_or_type="clear_output", content={"wait": wait}, ident=self.topic)
+        utils.get_kernel().iopub_send(msg_or_type="clear_output", content={"wait": wait}, ident=self.topic)
 
 
 class AsyncInteractiveShell(InteractiveShell):
@@ -136,7 +136,7 @@ class AsyncInteractiveShell(InteractiveShell):
     compile: Instance[XCachingCompiler]
     user_ns_hidden = Dict()
     _main_mod_cache = Dict()
-    _execute_request_timeout: ContextVar[float | None] = ContextVar("execute_request_timeout", default=None)
+
     run_cell = None  # pyright: ignore[reportAssignmentType]
     "**not-supported**"
     should_run_async = None  # pyright: ignore[reportAssignmentType]
@@ -162,21 +162,7 @@ def _default_banner1(self) -> str:
     @property
     def kernel(self) -> Kernel:
         "The current kernel."
-        return async_kernel.Kernel()
-
-    @property
-    def execute_request_timeout(self) -> float | None:
-        """A timeout in context of the [run_cell_async][async_kernel.asyncshell.AsyncInteractiveShell].
-
-        See also:
-
-        - [async_kernel.typing.MetadataKeys.timeout][].
-        """
-        return self._execute_request_timeout.get()
-
-    @execute_request_timeout.setter
-    def execute_request_timeout(self, value: float | None) -> None:
-        self._execute_request_timeout.set(value)
+        return utils.get_kernel()
 
     @observe("exit_now")
     def _update_exit_now(self, _) -> None:
@@ -250,7 +236,7 @@ async def run_cell_async(
         This function runs [execute requests][async_kernel.Kernel.execute_request] for the kernel
         wrapping [InteractiveShell][IPython.core.interactiveshell.InteractiveShell.run_cell_async].
         """
-        with anyio.fail_after(delay=self.execute_request_timeout):
+        with anyio.fail_after(delay=utils.get_execute_request_timeout()):
             result: ExecutionResult = await super().run_cell_async(
                 raw_cell=raw_cell,
                 store_history=store_history,
@@ -269,7 +255,7 @@ async def run_cell_async(
     def _showtraceback(self, etype, evalue, stb) -> None:
         if Tags.suppress_error in async_kernel.utils.get_tags():
             return
-        if self.execute_request_timeout is not None and etype is self.kernel.CancelledError:
+        if utils.get_execute_request_timeout() is not None and etype is self.kernel.CancelledError:
             etype, evalue, stb = TimeoutError, "Cell execute timeout", []
         self.kernel.iopub_send(
             msg_or_type="error",
@@ -294,14 +280,11 @@ class KernelMagics(Magics):
     @line_magic
     def connect_info(self, _) -> None:
         """Print information for connecting other clients to this kernel."""
-
-        kernel = async_kernel.Kernel()
+        kernel = utils.get_kernel()
         connection_file = pathlib.Path(kernel.connection_file)
-
         # if it's in the default dir, truncate to basename
         if jupyter_runtime_dir() == str(connection_file.parent):
             connection_file = connection_file.name
-
         info = kernel.get_connection_info()
         print(
             json.dumps(info, indent=2, default=json_default),

src/async_kernel/caller.py

@@ -238,8 +238,8 @@ class Caller:
     _pool_instances: ClassVar[weakref.WeakSet[Self]] = weakref.WeakSet()
     _executor_queue: dict
     _taskgroup: TaskGroup | None = None
-    _jobs: deque[tuple[contextvars.Context, tuple[Future, float, float, Callable, tuple, dict]] | Callable[[], Any]]
-    _jobs_added: threading.Event
+    _callers: deque[tuple[contextvars.Context, tuple[Future, float, float, Callable, tuple, dict]] | Callable[[], Any]]
+    _callers_added: threading.Event
     _stopped = False
     _protected = False
     _running = False
@@ -292,8 +292,8 @@ def __new__(
             inst.backend = Backend(sniffio.current_async_library())
             inst.thread = thread
             inst.log = log or logging.LoggerAdapter(logging.getLogger())
-            inst._jobs = deque()
-            inst._jobs_added = threading.Event()
+            inst._callers = deque()
+            inst._callers_added = threading.Event()
             inst._protected = protected
             inst._executor_queue = {}
             cls._instances[thread] = inst
@@ -325,8 +325,8 @@ async def _server_loop(self, tg: TaskGroup, task_status: TaskStatus[None]) -> No
             self.iopub_sockets[self.thread] = socket
             task_status.started()
             while not self._stopped:
-                while len(self._jobs):
-                    job = self._jobs.popleft()
+                while len(self._callers):
+                    job = self._callers.popleft()
                     if isinstance(job, Callable):
                         try:
                             job()
@@ -335,11 +335,11 @@ async def _server_loop(self, tg: TaskGroup, task_status: TaskStatus[None]) -> No
                     else:
                         context, args = job
                         context.run(tg.start_soon, self._wrap_call, *args)
-                    self._jobs_added.clear()
-                await wait_thread_event(self._jobs_added)
+                    self._callers_added.clear()
+                await wait_thread_event(self._callers_added)
         finally:
             self._running = False
-            for job in self._jobs:
+            for job in self._callers:
                 if not callable(job):
                     job[1][0].set_exception(FutureCancelledError())
             socket.close()
@@ -417,7 +417,7 @@ def stop(self, *, force=False) -> None:
         if self._protected and not force:
             return
         self._stopped = True
-        self._jobs_added.set()
+        self._callers_added.set()
         self._instances.pop(self.thread, None)
         if self in self._to_thread_pool:
             self._to_thread_pool.remove(self)
@@ -439,8 +439,8 @@ def call_later(
         if threading.current_thread() is self.thread and (tg := self._taskgroup):
             tg.start_soon(self._wrap_call, fut, time.monotonic(), delay, func, args, kwargs)
         else:
-            self._jobs.append((contextvars.copy_context(), (fut, time.monotonic(), delay, func, args, kwargs)))
-            self._jobs_added.set()
+            self._callers.append((contextvars.copy_context(), (fut, time.monotonic(), delay, func, args, kwargs)))
+            self._callers_added.set()
         self._outstanding += 1
         return fut
 
@@ -462,8 +462,8 @@ def call_no_context(self, func: Callable[P, Any], /, *args: P.args, **kwargs: P.
             *args: Arguments to use with func.
             **kwargs: Keyword arguments to use with func.
         """
-        self._jobs.append(functools.partial(func, *args, **kwargs))
-        self._jobs_added.set()
+        self._callers.append(functools.partial(func, *args, **kwargs))
+        self._callers_added.set()
 
     def has_execution_queue(self, func: Callable) -> bool:
         "Returns True if an execution queue exists for `func`."