asyncshell

Classes:

AsyncInteractiveShell –

An IPython InteractiveShell adapted to work with async-kernel.
AsyncInteractiveSubshell –

An asynchronous interactive subshell for managing isolated execution contexts within an async-kernel.
AsyncDisplayHook –

A displayhook subclass that publishes data using iopub_send.
AsyncDisplayPublisher –

A display publisher that publishes data using iopub_send.
SubshellManager –

Manages all instances of subshells.
KernelMagics –

Extra magics for async-kernel.

Attributes:

__all__ –

all `module-attribute` ¶

__all__ = ['AsyncInteractiveShell']

AsyncInteractiveShell ¶

Bases: InteractiveShell

An IPython InteractiveShell adapted to work with async-kernel.

Info

There is only one interactive shell instance.
When subclassing:
- The last defined subclass is used when creating the shell.
- A matching subclass of AsyncInteractiveSubshell should also be defined.

Notable differences

Supports a soft timeout specified via tags timeout=<value in seconds>¹.
user_ns and user_global_ns are same dictionary which is a fixed dict.
Supports async magic functions (See KernelMagics.pip).

When the execution time exceeds the timeout value, the code execution will "move on". ↩

Referenced by:

Reference
- kernel
  - Kernel
  - kernel
- typing
  - Tags
  - typing

Returned by:

Reference
- asyncshell SubshellManager get_shell
- kernel Kernel
  - main_shell
  - shell

Subclassed by:

Reference asyncshell AsyncInteractiveSubshell

Used by:

Reference kernel Kernel main_shell

Methods:

init_hooks –

Initialize hooks.
run_line_magic_async –

Call and awaits run_line_magic.
run_cell_magic_async –

Call and awaits run_cell_magic.
system –

Make a system call in a separate thread.
transform_cell_async –

Transform the cell and substitute magic calls with an awaitable wrapper.
execute_request –

Handle a execute request.
do_complete_request –

Handle a completion request.
is_complete_request –

Handle an is_complete request.
inspect_request –

Handle a inspect request.
history_request –

Handle a history request.
init_magics –

Initialize magics.
enable_matplotlib –

Enable interactive matplotlib and inline figure support.
context –

A context manager where the shell is active.
stop –

Stop the shell - do not call directly.

Attributes:

timeout –

A timeout in seconds to complete execute requests.
stop_on_error_time_offset –

An offset to add to the cancellation time to catch late arriving execute requests.

Source code in src/async_kernel/asyncshell.py

class AsyncInteractiveShell(InteractiveShell):
    """
    An IPython InteractiveShell adapted to work with [async-kernel][async_kernel.kernel.Kernel].

    Info:
        - There is only one interactive shell instance.
        - When subclassing:
            - The last defined subclass is used when creating the shell.
            - A matching subclass of [AsyncInteractiveSubshell][] should also be defined.

    Notable differences:
        - Supports a soft timeout specified via tags `timeout=<value in seconds>`[^1].
        - `user_ns` and `user_global_ns` are same dictionary which is a fixed [dict][].
        - Supports async magic functions (See [KernelMagics.pip][]).

        [^1]: When the execution time exceeds the timeout value, the code execution will "move on".
    """

    DEFAULT_MATPLOTLIB_BACKENDS = ["inline", "ipympl"]

    _execution_count = 0
    _resetting = False
    displayhook_class = traitlets.Type(AsyncDisplayHook)
    display_pub_class = traitlets.Type(AsyncDisplayPublisher)
    displayhook: traitlets.Instance[AsyncDisplayHook]
    display_pub: traitlets.Instance[AsyncDisplayPublisher]
    history_manager: HistoryManager
    compiler_class = traitlets.Type(XCachingCompiler)
    compile: traitlets.Instance[XCachingCompiler]
    kernel: traitlets.Instance[Kernel] = traitlets.Instance("async_kernel.Kernel", (), read_only=True)

    pending_manager = Fixed(ShellPendingManager)
    subshell_id = Fixed(lambda _: None)

    user_ns_hidden: Fixed[Self, dict] = Fixed(lambda c: c["owner"]._get_default_ns())
    user_global_ns: Fixed[Self, dict] = Fixed(lambda c: c["owner"]._user_ns)  # pyright: ignore[reportIncompatibleMethodOverride]

    _user_ns: Fixed[Self, dict] = Fixed(dict)  # pyright: ignore[reportIncompatibleVariableOverride]
    _main_mod_cache = Fixed(dict)
    _stop_on_error_pool: Fixed[Self, set[Callable[[], object]]] = Fixed(set)
    _stop_on_error_info: Fixed[Self, dict[Literal["time", "execution_count"], Any]] = Fixed(dict)

    timeout = traitlets.CFloat(0.0)
    "A timeout in seconds to complete execute requests."

    stop_on_error_time_offset = traitlets.Float(0.0)
    "An offset to add to the cancellation time to catch late arriving execute requests."

    loop_runner_map = None
    loop_runner = None
    autoindent = False

    @override
    def __repr__(self) -> str:
        return f"<{self.__class__.__name__}  kernel_name: {self.kernel.kernel_name!r} subshell_id: {self.subshell_id}>"

    def __new__(cls) -> Self:
        if issubclass(cls, AsyncInteractiveSubshell):
            return super().__new__(cls)
        if not InteractiveShell._instance:
            if cls is AsyncInteractiveShell._cls:
                return super().__new__(cls)
            InteractiveShell._instance = AsyncInteractiveShell._cls()  # pyright: ignore[reportAttributeAccessIssue]
            InteractiveShell.instance = SubshellManager.get_shell
        return InteractiveShell._instance  # pyright: ignore[reportReturnType]

    @override
    def __init__(self, parent=None) -> None:  # pyright: ignore[reportInconsistentConstructor]
        if not hasattr(self, "configurables"):
            super().__init__(parent=parent)

    def _get_default_ns(self) -> dict[str, Any]:
        # Copied from `InteractiveShell.init_user_ns`
        history = self.history_manager
        return {
            "_ih": getattr(history, "input_hist_parsed", False),
            "_oh": getattr(history, "output_hist", None),
            "_dh": getattr(history, "dir_hist", "."),
            "In": getattr(history, "input_hist_parsed", False),
            "Out": getattr(history, "output_hist", False),
            "get_ipython": self.get_ipython,
            "exit": self.exiter,
            "quit": self.exiter,
            "open": _modified_open_,
        }

    def __init_subclass__(cls) -> None:
        if AsyncInteractiveShell._instance:
            msg = "It is too late to define a subclass of AsyncInteractiveShell!"
            raise RuntimeError(msg)
        try:
            if not issubclass(cls, AsyncInteractiveSubshell):
                AsyncInteractiveShell._cls = cls
        except NameError:
            pass
        super().__init_subclass__()

    @traitlets.default("banner1")
    def _default_banner1(self) -> str:
        return (
            f"Python {sys.version}\n"
            f"async-kernel v{async_kernel.__version__}, {self.kernel.settings}) \n"
            f"IPython shell {IPython.core.release.version}\n"
        )

    @traitlets.observe("exit_now")
    def _update_exit_now(self, _) -> None:
        """Stop eventloop when `exit_now` fires."""
        if self.exit_now:
            self.kernel.interface.stop()

    def ask_exit(self) -> None:
        if self.kernel.interface.raw_input("Are you sure you want to stop the kernel?\ny/[n]\n") == "y":
            self.exit_now = True

    @override
    def init_create_namespaces(self, user_module=None, user_ns=None) -> None:
        return

    @override
    def save_sys_module_state(self) -> None:
        return

    @override
    def init_sys_modules(self) -> None:
        return

    @override
    def init_user_ns(self) -> None:
        return

    @override
    def init_hooks(self) -> None:
        """Initialize hooks."""
        super().init_hooks()

        def _show_in_pager(self, data: str | dict, start=0, screen_lines=0, pager_cmd=None) -> None:
            "Handle IPython page calls"
            if isinstance(data, dict):
                self.kernel.interface.iopub_send("display_data", content=data)
            else:
                self.kernel.interface.iopub_send("stream", content={"name": "stdout", "text": data})

        self.set_hook("show_in_pager", _show_in_pager, 99)

    @override
    def init_history(self) -> None:
        self.history_manager = HistoryManager(shell=self, parent=self)
        self.configurables.append(self.history_manager)  # pyright: ignore[reportArgumentType]
        utils.mark_thread_pydev_do_not_trace(self.history_manager.save_thread)

    @property
    @override
    def execution_count(self) -> int:
        return self._execution_count

    @execution_count.setter
    def execution_count(self, value) -> None:
        return

    @property
    @override
    def user_ns(self) -> dict[Any, Any]:
        ns = self._user_ns
        if "_ih" not in self._user_ns:
            ns.update(self._get_default_ns())
        return ns

    @user_ns.setter
    def user_ns(self, ns) -> None:
        ns = dict(ns)
        self.user_ns_hidden.clear()
        self._user_ns.clear()
        self.init_user_ns()
        ns_ = self._get_default_ns()
        self.user_ns_hidden.update(ns_)
        self._user_ns.update(ns_)
        self._user_ns.update(ns)

    @property
    @override
    def ns_table(self) -> dict[str, dict[Any, Any] | dict[str, Any]]:
        return {"user_global": self.user_global_ns, "user_local": self.user_ns, "builtin": builtins.__dict__}

    async def run_line_magic_async(self, magic_name: str, line: str, _stack_depth=1) -> Any:
        "Call and awaits [run_line_magic][IPython.core.interactiveshell.InteractiveShell.run_line_magic]."
        async with Caller().create_pending_group(mode=1):
            result = self.run_line_magic(magic_name, line, _stack_depth)
            try:
                return await result  # pyright: ignore[reportGeneralTypeIssues]
            except TypeError:
                return result

    async def run_cell_magic_async(self, magic_name: str, line: str, cell: str) -> Any:
        "Call and awaits [run_cell_magic][IPython.core.interactiveshell.InteractiveShell.run_cell_magic]."
        async with Caller().create_pending_group(mode=1):
            result = self.run_cell_magic(magic_name, line, cell)
            try:
                return await result  # pyright: ignore[reportGeneralTypeIssues]
            except TypeError:
                return result

    @override
    def system(self, cmd: list[str] | str, *, stderr_to_stdout: bool = False, **kwargs: Any) -> Pending[None]:
        """
        Make a system call in a separate thread.

        Args:
            cmd: Passed as the first argument 'command' when calling [anyio.open_process][].
            stderr_to_stdout: Send stderr output to stdout.
            **kwargs: Keyword arguments are passed to [anyio.open_process][].

        Tip:
            - The output can be redicted by making the call in the context of
                [async_kernel.utils.redirect_stdout][] and/or [async_kernel.utils.redirect_stderr][].
        """

        async def forward_output(transport_stream: ByteReceiveStream | None, out: TextIO, /) -> None:
            if transport_stream:
                async for text in TextReceiveStream(transport_stream):
                    out.write(text)

        async def open_process() -> None:
            async with await anyio.open_process(cmd, **kwargs) as process, anyio.create_task_group() as tg:
                tg.start_soon(forward_output, process.stdout, sys.stdout)
                tg.start_soon(forward_output, process.stderr, sys.stdout if stderr_to_stdout else sys.stderr)

        return Caller().to_thread(open_process)

    def transform_cell_async(self, raw_cell: str) -> str:
        "Transform the cell and substitute magic calls with an awaitable wrapper."

        return (
            self.transform_cell(raw_cell)
            .replace("get_ipython().run_line_magic(", "await get_ipython().run_line_magic_async(")
            .replace("get_ipython().run_cell_magic(", "await get_ipython().run_cell_magic_async(")
            .replace("get_ipython().system(", "await get_ipython().system(")
        )

    async def execute_request(
        self,
        code: str = "",
        *,
        silent: bool = False,
        store_history: bool = True,
        user_expressions: dict[str, str] | None = None,
        allow_stdin: bool = True,
        stop_on_error: bool = True,
        cell_id: str | None = None,
        received_time: float = 0,
        **_ignored,
    ) -> Content:
        """Handle a [execute request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#execute)."""
        if (received_time < self._stop_on_error_info.get("time", 0)) and not silent:
            return utils.error_to_content(RuntimeError("Aborting due to prior exception")) | {
                "execution_count": self._stop_on_error_info.get("execution_count", 0)
            }
        token = utils._cell_id_var.set(cell_id)  # pyright: ignore[reportPrivateUsage]
        try:
            tags: list[str] = utils.get_tags()
            timeout: float = utils.get_timeout(tags=tags)
            raises_exception: bool = Tags.raises_exception in tags
            stop_on_error_override: bool = Tags.stop_on_error in tags
            if stop_on_error_override:
                stop_on_error = utils.get_tag_value(Tags.stop_on_error, stop_on_error)
            elif raises_exception:
                stop_on_error = False

            if silent:
                execution_count: int = self.execution_count
            else:
                execution_count = self._execution_count = self._execution_count + 1
                self.kernel.interface.iopub_send(
                    msg_or_type="execute_input",
                    content={"code": code, "execution_count": execution_count},
                    ident=b"kernel.execute_input",
                )
            caller = Caller()
            err = None
            with anyio.CancelScope() as scope:

                def cancel():
                    if not silent:
                        caller.call_direct(scope.cancel, "Interrupted")

                result = None
                try:
                    self.kernel.interface.interrupts.add(cancel)
                    if stop_on_error:
                        self._stop_on_error_pool.add(cancel)
                    with anyio.fail_after(delay=timeout or None):
                        result = await self.run_cell_async(
                            raw_cell=code,
                            store_history=store_history,
                            silent=silent,
                            transformed_cell=self.transform_cell_async(code),
                            shell_futures=True,
                            cell_id=cell_id,
                        )
                except (Exception, anyio.get_cancelled_exc_class()) as e:
                    # A safeguard to catch exceptions not caught by the shell.
                    err = KernelInterrupt() if self.kernel.interface.last_interrupt_frame else e
                else:
                    err = result.error_before_exec or result.error_in_exec if result else KernelInterrupt()
                    if not err and Tags.raises_exception in tags:
                        msg = "An expected exception was not raised!"
                        err = RuntimeError(msg)
                finally:
                    self._stop_on_error_pool.discard(cancel)
                    self.kernel.interface.interrupts.discard(cancel)
                    self.events.trigger("post_execute")
                    if not silent:
                        self.events.trigger("post_run_cell", result)
            if (err) and (isinstance(err, anyio.get_cancelled_exc_class()) and (timeout != 0)):
                # Suppress the error due to either:
                # 1. tag
                # 2. timeout
                err = None
            content = {
                "status": "error" if err else "ok",
                "execution_count": execution_count,
                "user_expressions": self.user_expressions(user_expressions if user_expressions is not None else {}),
            }
            if err:
                content |= utils.error_to_content(err)
                if (not silent) and stop_on_error:
                    with anyio.CancelScope(shield=True):
                        await async_checkpoint(force=True)
                        self._stop_on_error_info["time"] = time.monotonic() + float(self.stop_on_error_time_offset)
                        self._stop_on_error_info["execution_count"] = execution_count
                        self.log.info("An error occurred in a non-silent execution request")
                        if stop_on_error:
                            for c in frozenset(self._stop_on_error_pool):
                                c()
            return content
        finally:
            utils._cell_id_var.reset(token)  # pyright: ignore[reportPrivateUsage]

    async def do_complete_request(self, code: str, cursor_pos: int | None = None) -> Content:
        """Handle a [completion request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#completion)."""

        cursor_pos = cursor_pos or len(code)
        with provisionalcompleter():
            completions = self.Completer.completions(code, cursor_pos)
            completions = list(rectify_completions(code, completions))
        comps = [
            {
                "start": comp.start,
                "end": comp.end,
                "text": comp.text,
                "type": comp.type,
                "signature": comp.signature,
            }
            for comp in completions
        ]
        s, e = (completions[0].start, completions[0].end) if completions else (cursor_pos, cursor_pos)
        matches = [c.text for c in completions]
        return {
            "matches": matches,
            "cursor_end": e,
            "cursor_start": s,
            "metadata": {"_jupyter_types_experimental": comps},
            "status": "ok",
        }

    async def is_complete_request(self, code: str) -> Content:
        """Handle an [is_complete request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#code-completeness)."""
        status, indent_spaces = self.input_transformer_manager.check_complete(code)
        content = {"status": status}
        if status == "incomplete":
            content["indent"] = " " * indent_spaces
        return content

    async def inspect_request(self, code: str, cursor_pos: int = 0, detail_level: Literal[0, 1] = 0) -> Content:
        """Handle a [inspect request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#introspection)."""
        content = {"data": {}, "metadata": {}, "found": True}
        try:
            oname = token_at_cursor(code, cursor_pos)
            bundle = self.object_inspect_mime(oname, detail_level=detail_level)
            content["data"] = bundle
        except KeyError:
            content["found"] = False
        return content

    async def history_request(
        self,
        *,
        output: bool = False,
        raw: bool = True,
        hist_access_type: str,
        session: int = 0,
        start: int = 1,
        stop: int | None = None,
        n: int = 10,
        pattern: str = "*",
        unique: bool = False,
        **_ignored,
    ) -> Content:
        """Handle a [history request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#history)."""
        history_manager = self.history_manager
        assert history_manager
        match hist_access_type:
            case "tail":
                hist = history_manager.get_tail(n=n, raw=raw, output=output, include_latest=False)
            case "range":
                hist = history_manager.get_range(session, start, stop, raw, output)
            case "search":
                hist = history_manager.search(pattern=pattern, raw=raw, output=output, n=n, unique=unique)
            case _:
                hist = []
        return {"history": list(hist), "status": "ok"}

    @override
    def _showtraceback(self, etype, evalue, stb) -> None:
        if utils.get_timeout() != 0.0 and etype is anyio.get_cancelled_exc_class():
            etype, evalue, stb = TimeoutError, "Cell execute timeout", []
        self.kernel.interface.iopub_send(
            msg_or_type="error",
            content={"traceback": stb, "ename": str(etype.__name__), "evalue": str(evalue)},
        )

    @override
    def reset(self, new_session=True, aggressive=False) -> None:
        if not self._resetting:
            self._resetting = True
            try:
                super().reset(new_session, aggressive)
                for pen in self.pending_manager.pending:
                    pen.cancel()
                if new_session:
                    self._execution_count = 0
                    self._stop_on_error_info.clear()
            finally:
                self._resetting = False

    @override
    def init_magics(self) -> None:
        """Initialize magics."""
        super().init_magics()
        self.register_magics(KernelMagics)
        self.magics_manager.register_alias("python", "thread")
        self.magics_manager.register_alias("python3", "thread")
        self.magics_manager.register_alias("!", "system")

    @override
    def enable_gui(self, gui=None) -> None:
        if (gui is not None) and (gui not in (supported := self._list_matplotlib_backends_and_gui_loops())):
            msg = f"The gui {gui!r} is not one of the supported gui options for this thread! {supported}="
            raise RuntimeError(msg)

    @override
    def enable_matplotlib(self, gui: str | None = None) -> tuple[str | Any | None, Any | str]:  # pragma: no cover
        """
        Enable interactive matplotlib and inline figure support.

        This takes the following steps:

        1. select the appropriate matplotlib backend
        2. set up matplotlib for interactive use with that backend
        3. configure formatters for inline figure display

        Args:
            gui:
                If given, dictates the choice of matplotlib GUI backend to use
                (should be one of IPython's supported backends, 'qt', 'osx', 'tk',
                'gtk', 'wx' or 'inline', 'ipympl'), otherwise we use the default chosen by
                matplotlib (as dictated by the matplotlib build-time options plus the
                user's matplotlibrc configuration file).  Note that not all backends
                make sense in all contexts, for example a terminal ipython can't
                display figures inline.
        """
        import matplotlib_inline.backend_inline  # noqa: PLC0415
        from IPython.core import pylabtools as pt  # noqa: PLC0415

        backends = self._list_matplotlib_backends_and_gui_loops()
        gui = gui or backends[0]
        gui, backend = pt.find_gui_and_backend(gui, self.pylab_gui_select)
        self.enable_gui(gui)
        try:
            pt.activate_matplotlib(backend)
        except RuntimeError as e:
            e.add_note(f"This thread supports the gui {gui!s} but pyplot only supports one interactive backend.")

        matplotlib_inline.backend_inline.configure_inline_support(self, backend)

        # Now we must activate the gui pylab wants to use, and fix %run to take
        # plot updates into account
        self.magics_manager.registry["ExecutionMagics"].default_runner = pt.mpl_runner(self.safe_execfile)

        return gui, backend

    def _list_matplotlib_backends_and_gui_loops(self) -> list[str | None]:
        return [*get_runtime_matplotlib_guis(), *self.DEFAULT_MATPLOTLIB_BACKENDS]

    @contextlib.contextmanager
    def context(self) -> Generator[None, Any, None]:
        "A context manager where the shell is active."
        with self.pending_manager.context():
            yield

    def stop(self, *, force=False) -> None:
        "Stop the shell - do not call directly."
        if force:
            self.reset(new_session=False)
            try:
                self.history_manager.end_session()
                self.history_manager.save_thread.stop()  # pyright: ignore[reportOptionalMemberAccess]
                self.history_manager.save_thread.join()  # pyright: ignore[reportOptionalMemberAccess]
            except AttributeError:
                pass
            SubshellManager.stop_all_subshells(force=True)
            InteractiveShell._instance = None

timeout `class-attribute` `instance-attribute` ¶

timeout = CFloat(0.0)

A timeout in seconds to complete execute requests.

stop_on_error_time_offset `class-attribute` `instance-attribute` ¶

stop_on_error_time_offset = Float(0.0)

An offset to add to the cancellation time to catch late arriving execute requests.

init_hooks ¶

init_hooks() -> None

Initialize hooks.

Source code in src/async_kernel/asyncshell.py

@override
def init_hooks(self) -> None:
    """Initialize hooks."""
    super().init_hooks()

    def _show_in_pager(self, data: str | dict, start=0, screen_lines=0, pager_cmd=None) -> None:
        "Handle IPython page calls"
        if isinstance(data, dict):
            self.kernel.interface.iopub_send("display_data", content=data)
        else:
            self.kernel.interface.iopub_send("stream", content={"name": "stdout", "text": data})

    self.set_hook("show_in_pager", _show_in_pager, 99)

run_line_magic_async `async` ¶

run_line_magic_async(magic_name: str, line: str, _stack_depth=1) -> Any

Call and awaits run_line_magic.

Source code in src/async_kernel/asyncshell.py

async def run_line_magic_async(self, magic_name: str, line: str, _stack_depth=1) -> Any:
    "Call and awaits [run_line_magic][IPython.core.interactiveshell.InteractiveShell.run_line_magic]."
    async with Caller().create_pending_group(mode=1):
        result = self.run_line_magic(magic_name, line, _stack_depth)
        try:
            return await result  # pyright: ignore[reportGeneralTypeIssues]
        except TypeError:
            return result

run_cell_magic_async `async` ¶

run_cell_magic_async(magic_name: str, line: str, cell: str) -> Any

Call and awaits run_cell_magic.

Source code in src/async_kernel/asyncshell.py

async def run_cell_magic_async(self, magic_name: str, line: str, cell: str) -> Any:
    "Call and awaits [run_cell_magic][IPython.core.interactiveshell.InteractiveShell.run_cell_magic]."
    async with Caller().create_pending_group(mode=1):
        result = self.run_cell_magic(magic_name, line, cell)
        try:
            return await result  # pyright: ignore[reportGeneralTypeIssues]
        except TypeError:
            return result

system ¶

system(
    cmd: list[str] | str, *, stderr_to_stdout: bool = False, **kwargs: Any
) -> Pending[None]

Make a system call in a separate thread.

Parameters:

cmd ¶
(list[str] | str) –

Passed as the first argument 'command' when calling anyio.open_process.
stderr_to_stdout ¶
(bool, default: False ) –

Send stderr output to stdout.
**kwargs ¶
(Any, default: {} ) –

Keyword arguments are passed to anyio.open_process.

Tip

The output can be redicted by making the call in the context of async_kernel.utils.redirect_stdout and/or async_kernel.utils.redirect_stderr.

Source code in src/async_kernel/asyncshell.py

@override
def system(self, cmd: list[str] | str, *, stderr_to_stdout: bool = False, **kwargs: Any) -> Pending[None]:
    """
    Make a system call in a separate thread.

    Args:
        cmd: Passed as the first argument 'command' when calling [anyio.open_process][].
        stderr_to_stdout: Send stderr output to stdout.
        **kwargs: Keyword arguments are passed to [anyio.open_process][].

    Tip:
        - The output can be redicted by making the call in the context of
            [async_kernel.utils.redirect_stdout][] and/or [async_kernel.utils.redirect_stderr][].
    """

    async def forward_output(transport_stream: ByteReceiveStream | None, out: TextIO, /) -> None:
        if transport_stream:
            async for text in TextReceiveStream(transport_stream):
                out.write(text)

    async def open_process() -> None:
        async with await anyio.open_process(cmd, **kwargs) as process, anyio.create_task_group() as tg:
            tg.start_soon(forward_output, process.stdout, sys.stdout)
            tg.start_soon(forward_output, process.stderr, sys.stdout if stderr_to_stdout else sys.stderr)

    return Caller().to_thread(open_process)

transform_cell_async ¶

transform_cell_async(raw_cell: str) -> str

Transform the cell and substitute magic calls with an awaitable wrapper.

Source code in src/async_kernel/asyncshell.py

def transform_cell_async(self, raw_cell: str) -> str:
    "Transform the cell and substitute magic calls with an awaitable wrapper."

    return (
        self.transform_cell(raw_cell)
        .replace("get_ipython().run_line_magic(", "await get_ipython().run_line_magic_async(")
        .replace("get_ipython().run_cell_magic(", "await get_ipython().run_cell_magic_async(")
        .replace("get_ipython().system(", "await get_ipython().system(")
    )

execute_request `async` ¶

execute_request(
    code: str = "",
    *,
    silent: bool = False,
    store_history: bool = True,
    user_expressions: dict[str, str] | None = None,
    allow_stdin: bool = True,
    stop_on_error: bool = True,
    cell_id: str | None = None,
    received_time: float = 0,
    **_ignored,
) -> Content

Handle a execute request.

Source code in src/async_kernel/asyncshell.py

async def execute_request(
    self,
    code: str = "",
    *,
    silent: bool = False,
    store_history: bool = True,
    user_expressions: dict[str, str] | None = None,
    allow_stdin: bool = True,
    stop_on_error: bool = True,
    cell_id: str | None = None,
    received_time: float = 0,
    **_ignored,
) -> Content:
    """Handle a [execute request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#execute)."""
    if (received_time < self._stop_on_error_info.get("time", 0)) and not silent:
        return utils.error_to_content(RuntimeError("Aborting due to prior exception")) | {
            "execution_count": self._stop_on_error_info.get("execution_count", 0)
        }
    token = utils._cell_id_var.set(cell_id)  # pyright: ignore[reportPrivateUsage]
    try:
        tags: list[str] = utils.get_tags()
        timeout: float = utils.get_timeout(tags=tags)
        raises_exception: bool = Tags.raises_exception in tags
        stop_on_error_override: bool = Tags.stop_on_error in tags
        if stop_on_error_override:
            stop_on_error = utils.get_tag_value(Tags.stop_on_error, stop_on_error)
        elif raises_exception:
            stop_on_error = False

        if silent:
            execution_count: int = self.execution_count
        else:
            execution_count = self._execution_count = self._execution_count + 1
            self.kernel.interface.iopub_send(
                msg_or_type="execute_input",
                content={"code": code, "execution_count": execution_count},
                ident=b"kernel.execute_input",
            )
        caller = Caller()
        err = None
        with anyio.CancelScope() as scope:

            def cancel():
                if not silent:
                    caller.call_direct(scope.cancel, "Interrupted")

            result = None
            try:
                self.kernel.interface.interrupts.add(cancel)
                if stop_on_error:
                    self._stop_on_error_pool.add(cancel)
                with anyio.fail_after(delay=timeout or None):
                    result = await self.run_cell_async(
                        raw_cell=code,
                        store_history=store_history,
                        silent=silent,
                        transformed_cell=self.transform_cell_async(code),
                        shell_futures=True,
                        cell_id=cell_id,
                    )
            except (Exception, anyio.get_cancelled_exc_class()) as e:
                # A safeguard to catch exceptions not caught by the shell.
                err = KernelInterrupt() if self.kernel.interface.last_interrupt_frame else e
            else:
                err = result.error_before_exec or result.error_in_exec if result else KernelInterrupt()
                if not err and Tags.raises_exception in tags:
                    msg = "An expected exception was not raised!"
                    err = RuntimeError(msg)
            finally:
                self._stop_on_error_pool.discard(cancel)
                self.kernel.interface.interrupts.discard(cancel)
                self.events.trigger("post_execute")
                if not silent:
                    self.events.trigger("post_run_cell", result)
        if (err) and (isinstance(err, anyio.get_cancelled_exc_class()) and (timeout != 0)):
            # Suppress the error due to either:
            # 1. tag
            # 2. timeout
            err = None
        content = {
            "status": "error" if err else "ok",
            "execution_count": execution_count,
            "user_expressions": self.user_expressions(user_expressions if user_expressions is not None else {}),
        }
        if err:
            content |= utils.error_to_content(err)
            if (not silent) and stop_on_error:
                with anyio.CancelScope(shield=True):
                    await async_checkpoint(force=True)
                    self._stop_on_error_info["time"] = time.monotonic() + float(self.stop_on_error_time_offset)
                    self._stop_on_error_info["execution_count"] = execution_count
                    self.log.info("An error occurred in a non-silent execution request")
                    if stop_on_error:
                        for c in frozenset(self._stop_on_error_pool):
                            c()
        return content
    finally:
        utils._cell_id_var.reset(token)  # pyright: ignore[reportPrivateUsage]

do_complete_request `async` ¶

do_complete_request(code: str, cursor_pos: int | None = None) -> Content

Handle a completion request.

Source code in src/async_kernel/asyncshell.py

async def do_complete_request(self, code: str, cursor_pos: int | None = None) -> Content:
    """Handle a [completion request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#completion)."""

    cursor_pos = cursor_pos or len(code)
    with provisionalcompleter():
        completions = self.Completer.completions(code, cursor_pos)
        completions = list(rectify_completions(code, completions))
    comps = [
        {
            "start": comp.start,
            "end": comp.end,
            "text": comp.text,
            "type": comp.type,
            "signature": comp.signature,
        }
        for comp in completions
    ]
    s, e = (completions[0].start, completions[0].end) if completions else (cursor_pos, cursor_pos)
    matches = [c.text for c in completions]
    return {
        "matches": matches,
        "cursor_end": e,
        "cursor_start": s,
        "metadata": {"_jupyter_types_experimental": comps},
        "status": "ok",
    }

is_complete_request `async` ¶

is_complete_request(code: str) -> Content

Handle an is_complete request.

Source code in src/async_kernel/asyncshell.py

async def is_complete_request(self, code: str) -> Content:
    """Handle an [is_complete request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#code-completeness)."""
    status, indent_spaces = self.input_transformer_manager.check_complete(code)
    content = {"status": status}
    if status == "incomplete":
        content["indent"] = " " * indent_spaces
    return content

inspect_request `async` ¶

inspect_request(
    code: str, cursor_pos: int = 0, detail_level: Literal[0, 1] = 0
) -> Content

Handle a inspect request.

Source code in src/async_kernel/asyncshell.py

async def inspect_request(self, code: str, cursor_pos: int = 0, detail_level: Literal[0, 1] = 0) -> Content:
    """Handle a [inspect request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#introspection)."""
    content = {"data": {}, "metadata": {}, "found": True}
    try:
        oname = token_at_cursor(code, cursor_pos)
        bundle = self.object_inspect_mime(oname, detail_level=detail_level)
        content["data"] = bundle
    except KeyError:
        content["found"] = False
    return content

history_request `async` ¶

history_request(
    *,
    output: bool = False,
    raw: bool = True,
    hist_access_type: str,
    session: int = 0,
    start: int = 1,
    stop: int | None = None,
    n: int = 10,
    pattern: str = "*",
    unique: bool = False,
    **_ignored,
) -> Content

Handle a history request.

Source code in src/async_kernel/asyncshell.py

async def history_request(
    self,
    *,
    output: bool = False,
    raw: bool = True,
    hist_access_type: str,
    session: int = 0,
    start: int = 1,
    stop: int | None = None,
    n: int = 10,
    pattern: str = "*",
    unique: bool = False,
    **_ignored,
) -> Content:
    """Handle a [history request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#history)."""
    history_manager = self.history_manager
    assert history_manager
    match hist_access_type:
        case "tail":
            hist = history_manager.get_tail(n=n, raw=raw, output=output, include_latest=False)
        case "range":
            hist = history_manager.get_range(session, start, stop, raw, output)
        case "search":
            hist = history_manager.search(pattern=pattern, raw=raw, output=output, n=n, unique=unique)
        case _:
            hist = []
    return {"history": list(hist), "status": "ok"}

init_magics ¶

init_magics() -> None

Initialize magics.

Source code in src/async_kernel/asyncshell.py

@override
def init_magics(self) -> None:
    """Initialize magics."""
    super().init_magics()
    self.register_magics(KernelMagics)
    self.magics_manager.register_alias("python", "thread")
    self.magics_manager.register_alias("python3", "thread")
    self.magics_manager.register_alias("!", "system")

enable_matplotlib ¶

enable_matplotlib(gui: str | None = None) -> tuple[str | Any | None, Any | str]

Enable interactive matplotlib and inline figure support.

This takes the following steps:

select the appropriate matplotlib backend
set up matplotlib for interactive use with that backend
configure formatters for inline figure display

Parameters:

gui ¶
(str | None, default: None ) –

If given, dictates the choice of matplotlib GUI backend to use (should be one of IPython's supported backends, 'qt', 'osx', 'tk', 'gtk', 'wx' or 'inline', 'ipympl'), otherwise we use the default chosen by matplotlib (as dictated by the matplotlib build-time options plus the user's matplotlibrc configuration file). Note that not all backends make sense in all contexts, for example a terminal ipython can't display figures inline.

Source code in src/async_kernel/asyncshell.py

@override
def enable_matplotlib(self, gui: str | None = None) -> tuple[str | Any | None, Any | str]:  # pragma: no cover
    """
    Enable interactive matplotlib and inline figure support.

    This takes the following steps:

    1. select the appropriate matplotlib backend
    2. set up matplotlib for interactive use with that backend
    3. configure formatters for inline figure display

    Args:
        gui:
            If given, dictates the choice of matplotlib GUI backend to use
            (should be one of IPython's supported backends, 'qt', 'osx', 'tk',
            'gtk', 'wx' or 'inline', 'ipympl'), otherwise we use the default chosen by
            matplotlib (as dictated by the matplotlib build-time options plus the
            user's matplotlibrc configuration file).  Note that not all backends
            make sense in all contexts, for example a terminal ipython can't
            display figures inline.
    """
    import matplotlib_inline.backend_inline  # noqa: PLC0415
    from IPython.core import pylabtools as pt  # noqa: PLC0415

    backends = self._list_matplotlib_backends_and_gui_loops()
    gui = gui or backends[0]
    gui, backend = pt.find_gui_and_backend(gui, self.pylab_gui_select)
    self.enable_gui(gui)
    try:
        pt.activate_matplotlib(backend)
    except RuntimeError as e:
        e.add_note(f"This thread supports the gui {gui!s} but pyplot only supports one interactive backend.")

    matplotlib_inline.backend_inline.configure_inline_support(self, backend)

    # Now we must activate the gui pylab wants to use, and fix %run to take
    # plot updates into account
    self.magics_manager.registry["ExecutionMagics"].default_runner = pt.mpl_runner(self.safe_execfile)

    return gui, backend

context ¶

context() -> Generator[None, Any, None]

A context manager where the shell is active.

Source code in src/async_kernel/asyncshell.py

@contextlib.contextmanager
def context(self) -> Generator[None, Any, None]:
    "A context manager where the shell is active."
    with self.pending_manager.context():
        yield

stop ¶

stop(*, force=False) -> None

Stop the shell - do not call directly.

Source code in src/async_kernel/asyncshell.py

def stop(self, *, force=False) -> None:
    "Stop the shell - do not call directly."
    if force:
        self.reset(new_session=False)
        try:
            self.history_manager.end_session()
            self.history_manager.save_thread.stop()  # pyright: ignore[reportOptionalMemberAccess]
            self.history_manager.save_thread.join()  # pyright: ignore[reportOptionalMemberAccess]
        except AttributeError:
            pass
        SubshellManager.stop_all_subshells(force=True)
        InteractiveShell._instance = None

AsyncInteractiveSubshell ¶

Bases: AsyncInteractiveShell

An asynchronous interactive subshell for managing isolated execution contexts within an async-kernel.

Each subshell has a unique user_ns, but shares its user_global_ns with the main shell (which is also the user_ns of the main shell).

Call subshell.stop(force=True) to stop a protected subshell when it is no longer required.

Attributes:

stopped –

Indicates whether the subshell has been stopped.
protected –

If True, prevents the subshell from being stopped unless forced.
pending_manager –

Tracks pending started in the context of the subshell.
subshell_id (Fixed[Self, str]) –

Unique identifier for the subshell.

Methods:

stop –

Stops the subshell, deactivating pending operations and removing it from the manager.

Info

The last defined subclass is used for all new subshells automatically.

stop ¶

stop(*, force=False) -> None

Stop this subshell.

Referenced by:

Reference asyncshell
- AsyncInteractiveSubshell
- SubshellManager
  - create_subshell
  - stop_all_subshells

Source code in src/async_kernel/asyncshell.py

@override
def stop(self, *, force=False) -> None:
    "Stop this subshell."
    if force or not self.protected:
        for pen in self.pending_manager.pending:
            pen.cancel(f"Subshell {self.subshell_id} is stopping.")
        self.reset(new_session=False)
        self.kernel.interface._subshell_stopped(self.subshell_id)  # pyright: ignore[reportPrivateUsage]
        SubshellManager.subshells.pop(self.subshell_id, None)
        self.set_trait("stopped", True)

AsyncDisplayHook ¶

Bases: DisplayHook

A displayhook subclass that publishes data using iopub_send.

Methods:

start_displayhook –

Start the display hook.
write_output_prompt –

Write the output prompt.
write_format_data –

Write format data to the message.
finish_displayhook –

Finish up all displayhook activities.

Source code in src/async_kernel/asyncshell.py

class AsyncDisplayHook(DisplayHook):
    """
    A displayhook subclass that publishes data using [iopub_send][async_kernel.interface.base.BaseKernelInterface.iopub_send].
    """

    shell: AsyncInteractiveShell
    _content: Fixed[Self, dict[int, dict[str, Any]]] = Fixed(dict)

    @property
    @override
    def prompt_count(self) -> int:
        return self.shell.execution_count

    @override
    def start_displayhook(self) -> None:
        """Start the display hook."""
        self._content[id(utils.get_job())] = {}

    @property
    def content(self) -> dict[str, Any]:
        return self._content[id(utils.get_job())]

    @override
    def write_output_prompt(self) -> None:
        """Write the output prompt."""
        self.content["execution_count"] = self.prompt_count

    @override
    def write_format_data(self, format_dict, md_dict=None) -> None:
        """Write format data to the message."""
        self.content["data"] = format_dict
        self.content["metadata"] = md_dict

    @override
    def finish_displayhook(self) -> None:
        """Finish up all displayhook activities."""
        if content := self.content:
            self.shell.kernel.interface.iopub_send("execute_result", content=content)
        self._content.pop(id(utils.get_job()))

start_displayhook ¶

start_displayhook() -> None

Start the display hook.

Source code in src/async_kernel/asyncshell.py

@override
def start_displayhook(self) -> None:
    """Start the display hook."""
    self._content[id(utils.get_job())] = {}

write_output_prompt ¶

write_output_prompt() -> None

Write the output prompt.

Source code in src/async_kernel/asyncshell.py

@override
def write_output_prompt(self) -> None:
    """Write the output prompt."""
    self.content["execution_count"] = self.prompt_count

write_format_data ¶

write_format_data(format_dict, md_dict=None) -> None

Write format data to the message.

Source code in src/async_kernel/asyncshell.py

@override
def write_format_data(self, format_dict, md_dict=None) -> None:
    """Write format data to the message."""
    self.content["data"] = format_dict
    self.content["metadata"] = md_dict

finish_displayhook ¶

finish_displayhook() -> None

Finish up all displayhook activities.

Source code in src/async_kernel/asyncshell.py

@override
def finish_displayhook(self) -> None:
    """Finish up all displayhook activities."""
    if content := self.content:
        self.shell.kernel.interface.iopub_send("execute_result", content=content)
    self._content.pop(id(utils.get_job()))

AsyncDisplayPublisher ¶

Bases: DisplayPublisher

A display publisher that publishes data using iopub_send.

Methods:

publish –

Publish a display-data message.
clear_output –

Clear output associated with the current execution (cell).
register_hook –

Register a hook for when publish is called.

Source code in src/async_kernel/asyncshell.py

class AsyncDisplayPublisher(DisplayPublisher):
    """A display publisher that publishes data using [iopub_send][async_kernel.interface.base.BaseKernelInterface.iopub_send]."""

    shell: AsyncInteractiveShell
    _hooks: Fixed[Self, list[Callable[[Message[Any]], Any]]] = Fixed(list)

    @override
    def publish(  # pyright: ignore[reportIncompatibleMethodOverride]
        self,
        data: dict[str, Any],
        metadata: dict | None = None,
        *,
        transient: dict | None = None,
        update: bool = False,
        **kwargs,
    ) -> None:
        """
        Publish a display-data message.

        Args:
            data: A mime-bundle dict, keyed by mime-type.
            metadata: Metadata associated with the data.
            transient: Transient data that may only be relevant during a live display, such as display_id.
                Transient data should not be persisted to documents.
            update: If True, send an update_display_data message instead of display_data.

        [Reference](https://jupyter-client.readthedocs.io/en/stable/messaging.html#update-display-data)
        """
        content = {"data": data, "metadata": metadata or {}, "transient": transient or {}} | kwargs
        msg_type = "update_display_data" if update else "display_data"
        msg = self.shell.kernel.interface.msg(msg_type, content=content, parent=utils.get_parent())
        for hook in self._hooks:
            try:
                msg = hook(msg)
            except Exception:
                pass
            if msg is None:
                return
        if "application/vnd.jupyter.widget-view+json" in data and os.environ.get("VSCODE_CWD"):  # pragma: no cover
            # ref: https://github.com/microsoft/vscode-jupyter/wiki/Component:-IPyWidgets#two-widget-managers
            # On occasion we get `Error 'widget model not found'`
            # As a work-around inject a delay so the widget can be registered first.
            self.shell.kernel.interface.callers[Channel.control].call_later(
                0.2, self.shell.kernel.interface.iopub_send, msg
            )
        else:
            self.shell.kernel.interface.iopub_send(msg)

    @override
    def clear_output(self, wait: bool = False) -> None:
        """
        Clear output associated with the current execution (cell).

        Args:
            wait: If True, the output will not be cleared immediately,
                instead waiting for the next display before clearing.
                This reduces bounce during repeated clear & display loops.
        """
        self.shell.kernel.interface.iopub_send(
            msg_or_type="clear_output", content={"wait": wait}, ident=b"display_data"
        )

    def register_hook(self, hook: Callable[[Message[Any]], Any]) -> None:
        """Register a hook for when publish is called.

        The hook should return the message or None.
        Only return `None` when the message should *not* be sent.
        """
        self._hooks.append(hook)

    def unregister_hook(self, hook: Callable[[Message[Any]], Any]) -> None:
        while hook in self._hooks:
            self._hooks.remove(hook)

publish ¶

publish(
    data: dict[str, Any],
    metadata: dict | None = None,
    *,
    transient: dict | None = None,
    update: bool = False,
    **kwargs,
) -> None

Publish a display-data message.

Parameters:

data ¶
(dict[str, Any]) –

A mime-bundle dict, keyed by mime-type.
metadata ¶
(dict | None, default: None ) –

Metadata associated with the data.
transient ¶
(dict | None, default: None ) –

Transient data that may only be relevant during a live display, such as display_id. Transient data should not be persisted to documents.
update ¶
(bool, default: False ) –

If True, send an update_display_data message instead of display_data.

Reference

Source code in src/async_kernel/asyncshell.py

@override
def publish(  # pyright: ignore[reportIncompatibleMethodOverride]
    self,
    data: dict[str, Any],
    metadata: dict | None = None,
    *,
    transient: dict | None = None,
    update: bool = False,
    **kwargs,
) -> None:
    """
    Publish a display-data message.

    Args:
        data: A mime-bundle dict, keyed by mime-type.
        metadata: Metadata associated with the data.
        transient: Transient data that may only be relevant during a live display, such as display_id.
            Transient data should not be persisted to documents.
        update: If True, send an update_display_data message instead of display_data.

    [Reference](https://jupyter-client.readthedocs.io/en/stable/messaging.html#update-display-data)
    """
    content = {"data": data, "metadata": metadata or {}, "transient": transient or {}} | kwargs
    msg_type = "update_display_data" if update else "display_data"
    msg = self.shell.kernel.interface.msg(msg_type, content=content, parent=utils.get_parent())
    for hook in self._hooks:
        try:
            msg = hook(msg)
        except Exception:
            pass
        if msg is None:
            return
    if "application/vnd.jupyter.widget-view+json" in data and os.environ.get("VSCODE_CWD"):  # pragma: no cover
        # ref: https://github.com/microsoft/vscode-jupyter/wiki/Component:-IPyWidgets#two-widget-managers
        # On occasion we get `Error 'widget model not found'`
        # As a work-around inject a delay so the widget can be registered first.
        self.shell.kernel.interface.callers[Channel.control].call_later(
            0.2, self.shell.kernel.interface.iopub_send, msg
        )
    else:
        self.shell.kernel.interface.iopub_send(msg)

clear_output ¶

clear_output(wait: bool = False) -> None

Clear output associated with the current execution (cell).

Parameters:

wait ¶
(bool, default: False ) –

If True, the output will not be cleared immediately, instead waiting for the next display before clearing. This reduces bounce during repeated clear & display loops.

Source code in src/async_kernel/asyncshell.py

@override
def clear_output(self, wait: bool = False) -> None:
    """
    Clear output associated with the current execution (cell).

    Args:
        wait: If True, the output will not be cleared immediately,
            instead waiting for the next display before clearing.
            This reduces bounce during repeated clear & display loops.
    """
    self.shell.kernel.interface.iopub_send(
        msg_or_type="clear_output", content={"wait": wait}, ident=b"display_data"
    )

register_hook ¶

register_hook(hook: Callable[[Message[Any]], Any]) -> None

Register a hook for when publish is called.

The hook should return the message or None. Only return None when the message should not be sent.

Source code in src/async_kernel/asyncshell.py

def register_hook(self, hook: Callable[[Message[Any]], Any]) -> None:
    """Register a hook for when publish is called.

    The hook should return the message or None.
    Only return `None` when the message should *not* be sent.
    """
    self._hooks.append(hook)

SubshellManager ¶

Manages all instances of subshells.

Note:

- All methods are [classmethod][].

Used by:

Reference kernel Kernel subshell_manager

Methods:

create_subshell –

Create a new instance of the default subshell class.
get_shell –

Get a subshell or the main shell.
delete_subshell –

Stop a subshell unless it is protected.
stop_all_subshells –

Stop all current subshells.

Source code in src/async_kernel/asyncshell.py

@final
class SubshellManager:
    """
    Manages all instances of [subshells][async_kernel.asyncshell.AsyncInteractiveSubshell].

    Note:

        - All methods are [classmethod][].
    """

    subshells: dict[str, AsyncInteractiveSubshell] = {}

    def __new__(cls) -> Never:
        msg = "Instantiation is not required, use classmethods directly."
        raise RuntimeError(msg)

    @classmethod
    def create_subshell(cls, *, protected: bool = True) -> AsyncInteractiveSubshell:
        """
        Create a new instance of the default subshell class.

        Call [`subshell.stop(force=True)`][async_kernel.asyncshell.AsyncInteractiveSubshell.stop] to stop a
        protected subshell when it is no longer required.

        Args:
            protected: Protect the subshell from accidental deletion.
        """
        return AsyncInteractiveSubshell(protected=protected)

    @classmethod
    def list_subshells(cls) -> list[str]:
        return list(cls.subshells)

    @classmethod
    def get_shell(
        cls,
        subshell_id: str | None | NoValue = NoValue,  # pyright: ignore[reportInvalidTypeForm]
    ) -> AsyncInteractiveShell | AsyncInteractiveSubshell:
        """
        Get a subshell or the main shell.

        Args:
            subshell_id: The id of an existing subshell.
        """
        mainshell = AsyncInteractiveShell()
        if subshell_id is NoValue:
            subshell_id = ShellPendingManager.active_id()
        if subshell_id is None or subshell_id == mainshell.pending_manager.id:
            return mainshell
        return cls.subshells[subshell_id]

    @classmethod
    def delete_subshell(cls, subshell_id: str) -> None:
        """
        Stop a subshell unless it is protected.

        Args:
            subshell_id: The id of an existing subshell to stop.
        """
        if subshell := cls.subshells.get(subshell_id):
            subshell.stop()

    @classmethod
    def stop_all_subshells(cls, *, force: bool = False) -> None:
        """Stop all current subshells.

        Args:
            force: Passed to [async_kernel.asyncshell.AsyncInteractiveSubshell.stop][].
        """
        for subshell in set(cls.subshells.values()):
            subshell.stop(force=force)

create_subshell `classmethod` ¶

create_subshell(*, protected: bool = True) -> AsyncInteractiveSubshell

Create a new instance of the default subshell class.

Call subshell.stop(force=True) to stop a protected subshell when it is no longer required.

Parameters:

protected ¶
(bool, default: True ) –

Protect the subshell from accidental deletion.

Source code in src/async_kernel/asyncshell.py

@classmethod
def create_subshell(cls, *, protected: bool = True) -> AsyncInteractiveSubshell:
    """
    Create a new instance of the default subshell class.

    Call [`subshell.stop(force=True)`][async_kernel.asyncshell.AsyncInteractiveSubshell.stop] to stop a
    protected subshell when it is no longer required.

    Args:
        protected: Protect the subshell from accidental deletion.
    """
    return AsyncInteractiveSubshell(protected=protected)

get_shell `classmethod` ¶

get_shell(
    subshell_id: str | None | NoValue = NoValue,
) -> AsyncInteractiveShell | AsyncInteractiveSubshell

Get a subshell or the main shell.

Parameters:

subshell_id ¶
(str | None | NoValue, default: NoValue ) –

The id of an existing subshell.

Source code in src/async_kernel/asyncshell.py

@classmethod
def get_shell(
    cls,
    subshell_id: str | None | NoValue = NoValue,  # pyright: ignore[reportInvalidTypeForm]
) -> AsyncInteractiveShell | AsyncInteractiveSubshell:
    """
    Get a subshell or the main shell.

    Args:
        subshell_id: The id of an existing subshell.
    """
    mainshell = AsyncInteractiveShell()
    if subshell_id is NoValue:
        subshell_id = ShellPendingManager.active_id()
    if subshell_id is None or subshell_id == mainshell.pending_manager.id:
        return mainshell
    return cls.subshells[subshell_id]

delete_subshell `classmethod` ¶

delete_subshell(subshell_id: str) -> None

Stop a subshell unless it is protected.

Parameters:

subshell_id ¶
(str) –

The id of an existing subshell to stop.

Source code in src/async_kernel/asyncshell.py

@classmethod
def delete_subshell(cls, subshell_id: str) -> None:
    """
    Stop a subshell unless it is protected.

    Args:
        subshell_id: The id of an existing subshell to stop.
    """
    if subshell := cls.subshells.get(subshell_id):
        subshell.stop()

stop_all_subshells `classmethod` ¶

stop_all_subshells(*, force: bool = False) -> None

Stop all current subshells.

Parameters:

force ¶
(bool, default: False ) –

Passed to async_kernel.asyncshell.AsyncInteractiveSubshell.stop.

Source code in src/async_kernel/asyncshell.py

@classmethod
def stop_all_subshells(cls, *, force: bool = False) -> None:
    """Stop all current subshells.

    Args:
        force: Passed to [async_kernel.asyncshell.AsyncInteractiveSubshell.stop][].
    """
    for subshell in set(cls.subshells.values()):
        subshell.stop(force=force)

KernelMagics ¶

Bases: Magics

Extra magics for async-kernel.

Methods:

connect_info –

Print information for connecting other clients to this kernel.
callers –

Print a table of Callers indicating it's status.
subshell –

Print subshell info ref.
pip –

Run the pip package manager for the current environment.
uv –

Run the uv package manager for the current environment.
thread –

Run the python code in a caller managed child thread.
asyncio –
trio –

Source code in src/async_kernel/asyncshell.py

@magics_class
class KernelMagics(Magics):
    """Extra magics for async-kernel."""

    shell: AsyncInteractiveShell  # pyright: ignore[reportIncompatibleVariableOverride]

    @line_magic
    def connect_info(self, _) -> None:
        """Print information for connecting other clients to this kernel."""
        connection_file = pathlib.Path(self.shell.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 = self.shell.kernel.get_connection_info()
        print(
            json.dumps(info, indent=2),
            "Paste the above JSON into a file, and connect with:\n"
            + "    $> jupyter <app> --existing <file>\n"
            + "or, if you are local, you can connect with just:\n"
            + f"    $> jupyter <app> --existing {connection_file}\n"
            + "or even just:\n"
            + "    $> jupyter <app> --existing\n"
            + "if this is the most recent Jupyter kernel you have started.",
        )

    @line_magic
    def callers(self, _) -> None:
        "Print a table of [Callers][async_kernel.caller.Caller] indicating it's status."
        callers = Caller.all_callers(running_only=False)
        n = max(len(c.name) for c in callers) + 6
        m = max(len(repr(c.id)) for c in callers) + 6
        t = max(len(str(c.thread.name)) for c in callers) + 6
        lines = [
            "".join(["Name".center(n), "Running ", "Protected", "Thread".center(t), "Caller".center(m)]),
            "─" * (n + m + t + 22),
        ]
        for caller in callers:
            running = ("✓" if caller.running else "✗").center(8)
            protected = "   🔐    " if caller.protected else "         "
            name = caller.name + " " * (n - len(caller.name))
            thread = str(caller.thread.name).center(t)
            caller_id = str(caller.id)
            if caller.id == Caller.id_current():
                caller_id += " ← current"
            lines.append("".join([name, running.center(8), protected, thread, caller_id]))
        print(*lines, sep="\n")

    @line_magic
    def subshell(self, _) -> None:
        """
        Print subshell info [ref](https://jupyter.org/enhancement-proposals/91-kernel-subshells/kernel-subshells.html#list-subshells).
        """
        subshells = SubshellManager.list_subshells()
        subshell_list = (
            f"\t----- {len(subshells)} x subshell -----\n" + "\n".join(subshells) if subshells else "-- No subshells --"
        )
        print(f"Current shell:\t{self.shell}\n\n{subshell_list}")

    @no_var_expand
    @line_magic
    async def pip(self, line: str) -> Any | None:
        """Run the pip package manager for the current environment.

        Usage:
          %pip install [pkgs]
        """
        if sys.platform == "emscripten":
            import micropip  # noqa: PLC0415

            match line.split(maxsplit=1)[0]:
                case "install":
                    requirements = [
                        f"emfs:{n}" if n.startswith(".") else n for n in line.removeprefix("install").split()
                    ]
                    return await micropip.install(requirements, verbose=True)
                case "uninstall":
                    return micropip.uninstall(line.removeprefix("uninstall").split(), verbose=True)
                case "freeze":
                    return micropip.freeze()
                case "list":
                    return micropip.list()
                case _ as name:
                    print("Unsupported command:", name)
        else:
            await SubshellManager.get_shell().system([sys.executable, "-m", "pip", *line.split()])
        return None

    @no_var_expand
    @line_magic
    def uv(self, line) -> Pending[None]:
        """Run the uv package manager for the current environment.

        Usage:
          %uv pip install [pkgs]
        """
        return SubshellManager.get_shell().system(["uv", *shlex.split(line)], stderr_to_stdout=True)

    @no_var_expand
    @line_cell_magic
    async def thread(self, line: str, cell: str | None = None) -> None:
        """
        Run the python code in a caller managed child thread.

        Both line and cell magic are supported.

        For cell_magic, [CallerCreateOptions][async_kernel.typing.CallerCreateOptions] can be passed as literals.

        Example:
            %%thread name="Trio executor" backend=trio
        """
        shell: AsyncInteractiveShell | AsyncInteractiveSubshell = SubshellManager.get_shell()
        if cell is None:
            cell = line
            options: Any = None
        else:
            options = RunMode.line_to_options(line)
        caller = shell.kernel.caller
        await (caller.get(**options).call_soon if options else caller.to_thread)(
            shell.run_cell_async,
            raw_cell=cell,
            store_history=False,
            silent=True,
            cell_id=None,
            transformed_cell=shell.transform_cell_async(cell),
        )

    async def _call_using_backend(self, backend: Literal["asyncio", "trio"], code: str) -> None:
        shell: AsyncInteractiveShell | AsyncInteractiveSubshell = SubshellManager.get_shell()
        await Caller().call_using_backend(
            backend,
            shell.run_cell_async,
            raw_cell=code,
            store_history=False,
            silent=True,
            cell_id=None,
            transformed_cell=shell.transform_cell_async(code),
        )

    @no_var_expand
    @line_cell_magic
    async def asyncio(self, line: str, cell: str | None = None) -> None:
        ""
        await self._call_using_backend("asyncio", cell or line)

    @no_var_expand
    @line_cell_magic
    async def trio(self, line: str, cell: str | None = None) -> None:
        ""
        await self._call_using_backend("trio", cell or line)

connect_info ¶

connect_info(_) -> None

Print information for connecting other clients to this kernel.

Source code in src/async_kernel/asyncshell.py

@line_magic
def connect_info(self, _) -> None:
    """Print information for connecting other clients to this kernel."""
    connection_file = pathlib.Path(self.shell.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 = self.shell.kernel.get_connection_info()
    print(
        json.dumps(info, indent=2),
        "Paste the above JSON into a file, and connect with:\n"
        + "    $> jupyter <app> --existing <file>\n"
        + "or, if you are local, you can connect with just:\n"
        + f"    $> jupyter <app> --existing {connection_file}\n"
        + "or even just:\n"
        + "    $> jupyter <app> --existing\n"
        + "if this is the most recent Jupyter kernel you have started.",
    )

callers ¶

callers(_) -> None

Print a table of Callers indicating it's status.

Source code in src/async_kernel/asyncshell.py

@line_magic
def callers(self, _) -> None:
    "Print a table of [Callers][async_kernel.caller.Caller] indicating it's status."
    callers = Caller.all_callers(running_only=False)
    n = max(len(c.name) for c in callers) + 6
    m = max(len(repr(c.id)) for c in callers) + 6
    t = max(len(str(c.thread.name)) for c in callers) + 6
    lines = [
        "".join(["Name".center(n), "Running ", "Protected", "Thread".center(t), "Caller".center(m)]),
        "─" * (n + m + t + 22),
    ]
    for caller in callers:
        running = ("✓" if caller.running else "✗").center(8)
        protected = "   🔐    " if caller.protected else "         "
        name = caller.name + " " * (n - len(caller.name))
        thread = str(caller.thread.name).center(t)
        caller_id = str(caller.id)
        if caller.id == Caller.id_current():
            caller_id += " ← current"
        lines.append("".join([name, running.center(8), protected, thread, caller_id]))
    print(*lines, sep="\n")

subshell ¶

subshell(_) -> None

Print subshell info ref.

Source code in src/async_kernel/asyncshell.py

@line_magic
def subshell(self, _) -> None:
    """
    Print subshell info [ref](https://jupyter.org/enhancement-proposals/91-kernel-subshells/kernel-subshells.html#list-subshells).
    """
    subshells = SubshellManager.list_subshells()
    subshell_list = (
        f"\t----- {len(subshells)} x subshell -----\n" + "\n".join(subshells) if subshells else "-- No subshells --"
    )
    print(f"Current shell:\t{self.shell}\n\n{subshell_list}")

pip `async` ¶

pip(line: str) -> Any | None

Run the pip package manager for the current environment.

Usage

%pip install [pkgs]

Referenced by:

Reference asyncshell AsyncInteractiveShell

Source code in src/async_kernel/asyncshell.py

@no_var_expand
@line_magic
async def pip(self, line: str) -> Any | None:
    """Run the pip package manager for the current environment.

    Usage:
      %pip install [pkgs]
    """
    if sys.platform == "emscripten":
        import micropip  # noqa: PLC0415

        match line.split(maxsplit=1)[0]:
            case "install":
                requirements = [
                    f"emfs:{n}" if n.startswith(".") else n for n in line.removeprefix("install").split()
                ]
                return await micropip.install(requirements, verbose=True)
            case "uninstall":
                return micropip.uninstall(line.removeprefix("uninstall").split(), verbose=True)
            case "freeze":
                return micropip.freeze()
            case "list":
                return micropip.list()
            case _ as name:
                print("Unsupported command:", name)
    else:
        await SubshellManager.get_shell().system([sys.executable, "-m", "pip", *line.split()])
    return None

uv ¶

uv(line) -> Pending[None]

Run the uv package manager for the current environment.

Usage

%uv pip install [pkgs]

Source code in src/async_kernel/asyncshell.py

@no_var_expand
@line_magic
def uv(self, line) -> Pending[None]:
    """Run the uv package manager for the current environment.

    Usage:
      %uv pip install [pkgs]
    """
    return SubshellManager.get_shell().system(["uv", *shlex.split(line)], stderr_to_stdout=True)

thread `async` ¶

thread(line: str, cell: str | None = None) -> None

Run the python code in a caller managed child thread.

Both line and cell magic are supported.

For cell_magic, CallerCreateOptions can be passed as literals.

Example

%%thread name="Trio executor" backend=trio

Source code in src/async_kernel/asyncshell.py

@no_var_expand
@line_cell_magic
async def thread(self, line: str, cell: str | None = None) -> None:
    """
    Run the python code in a caller managed child thread.

    Both line and cell magic are supported.

    For cell_magic, [CallerCreateOptions][async_kernel.typing.CallerCreateOptions] can be passed as literals.

    Example:
        %%thread name="Trio executor" backend=trio
    """
    shell: AsyncInteractiveShell | AsyncInteractiveSubshell = SubshellManager.get_shell()
    if cell is None:
        cell = line
        options: Any = None
    else:
        options = RunMode.line_to_options(line)
    caller = shell.kernel.caller
    await (caller.get(**options).call_soon if options else caller.to_thread)(
        shell.run_cell_async,
        raw_cell=cell,
        store_history=False,
        silent=True,
        cell_id=None,
        transformed_cell=shell.transform_cell_async(cell),
    )

asyncio `async` ¶

asyncio(line: str, cell: str | None = None) -> None

Source code in src/async_kernel/asyncshell.py

@no_var_expand
@line_cell_magic
async def asyncio(self, line: str, cell: str | None = None) -> None:
    ""
    await self._call_using_backend("asyncio", cell or line)

trio `async` ¶

trio(line: str, cell: str | None = None) -> None

Source code in src/async_kernel/asyncshell.py

@no_var_expand
@line_cell_magic
async def trio(self, line: str, cell: str | None = None) -> None:
    ""
    await self._call_using_backend("trio", cell or line)

asyncshell

__all__ module-attribute ¶

AsyncInteractiveShell ¶

timeout class-attribute instance-attribute ¶

stop_on_error_time_offset class-attribute instance-attribute ¶

init_hooks ¶

run_line_magic_async async ¶

run_cell_magic_async async ¶

system ¶

cmd ¶

stderr_to_stdout ¶

**kwargs ¶

transform_cell_async ¶

execute_request async ¶

do_complete_request async ¶

is_complete_request async ¶

inspect_request async ¶

history_request async ¶

init_magics ¶

enable_matplotlib ¶

gui ¶

context ¶

stop ¶

AsyncInteractiveSubshell ¶

stop ¶

AsyncDisplayHook ¶

start_displayhook ¶

write_output_prompt ¶

write_format_data ¶

finish_displayhook ¶

AsyncDisplayPublisher ¶

publish ¶

data ¶

metadata ¶

transient ¶

update ¶

clear_output ¶

wait ¶

register_hook ¶

SubshellManager ¶

create_subshell classmethod ¶

protected ¶

get_shell classmethod ¶

subshell_id ¶

delete_subshell classmethod ¶

subshell_id ¶

stop_all_subshells classmethod ¶

force ¶

KernelMagics ¶

connect_info ¶

callers ¶

subshell ¶

pip async ¶

uv ¶

thread async ¶

asyncio async ¶

trio async ¶

all `module-attribute` ¶

timeout `class-attribute` `instance-attribute` ¶

stop_on_error_time_offset `class-attribute` `instance-attribute` ¶

run_line_magic_async `async` ¶

run_cell_magic_async `async` ¶

`cmd` ¶

`stderr_to_stdout` ¶

`kwargs`** ¶

execute_request `async` ¶

do_complete_request `async` ¶

is_complete_request `async` ¶

inspect_request `async` ¶

history_request `async` ¶

`gui` ¶

`data` ¶

`metadata` ¶

`transient` ¶

`update` ¶

`wait` ¶

create_subshell `classmethod` ¶

`protected` ¶

get_shell `classmethod` ¶

`subshell_id` ¶

delete_subshell `classmethod` ¶

`subshell_id` ¶

stop_all_subshells `classmethod` ¶

`force` ¶

pip `async` ¶

thread `async` ¶

asyncio `async` ¶

trio `async` ¶