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.
KernelInterruptError –

Raised to interrupt the 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.

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.

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 shell

Subclassed by:

Reference asyncshell AsyncInteractiveSubshell

Methods:

init_hooks –

Initialize hooks.
init_magics –

Initialize magics.
enable_matplotlib –

Enable interactive matplotlib and inline figure support.
context –

A context manager where the shell is active.

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].

    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][].

        [^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 = Type(AsyncDisplayHook)
    display_pub_class = Type(AsyncDisplayPublisher)
    displayhook: Instance[AsyncDisplayHook]
    display_pub: Instance[AsyncDisplayPublisher]
    compiler_class = Type(XCachingCompiler)
    compile: Instance[XCachingCompiler]
    kernel: Instance[Kernel] = 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 = CFloat(0.0)
    "A timeout in seconds to complete execute requests."

    stop_on_error_time_offset = 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}>"

    @override
    def __init__(self, parent: None | Configurable = None) -> None:
        super().__init__(parent=parent)
        with contextlib.suppress(AttributeError):
            utils.mark_thread_pydev_do_not_trace(self.history_manager.save_thread)  # pyright: ignore[reportOptionalMemberAccess]

    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_,
        }

    @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"
        )

    @observe("exit_now")
    def _update_exit_now(self, _) -> None:
        """Stop eventloop when `exit_now` fires."""
        if self.exit_now:
            self.kernel.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)

    @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 _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)
            suppress_error: bool = Tags.suppress_error in 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 suppress_error or 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.iopub_send(
                    msg_or_type="execute_input",
                    content={"code": code, "execution_count": execution_count},
                    ident=self.kernel.topic("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(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 = KernelInterruptError() if self.kernel.interface.last_interrupt_frame else e
                else:
                    err = result.error_before_exec or result.error_in_exec if result else KernelInterruptError()
                    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 (suppress_error or (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: HistoryManager = self.history_manager  # pyright: ignore[reportAssignmentType]
        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 Tags.suppress_error in utils.get_tags():
            if msg := utils.get_tag_value(Tags.suppress_error, "⚠"):
                print(msg)
            return
        if utils.get_timeout() != 0.0 and etype is anyio.get_cancelled_exc_class():
            etype, evalue, stb = TimeoutError, "Cell execute timeout", []
        self.kernel.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)

    @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

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)

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)

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

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.

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

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._subshell_stopped(self.subshell_id)  # pyright: ignore[reportPrivateUsage]
        self.kernel.subshell_manager.subshells.pop(self.subshell_id, None)
        self.set_trait("stopped", True)

KernelInterruptError ¶

Bases: Exception

Raised to interrupt the kernel.

Referenced by:

Reference interface
- BaseKernelInterface interrupt
- BaseKernelInterface

Source code in src/async_kernel/asyncshell.py

class KernelInterruptError(Exception):
    "Raised to interrupt the kernel."

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.kernel.Kernel.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.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.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.kernel.Kernel.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.callers[Channel.control].call_later(0.2, self.shell.kernel.iopub_send, msg)
        else:
            self.shell.kernel.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.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.callers[Channel.control].call_later(0.2, self.shell.kernel.iopub_send, msg)
    else:
        self.shell.kernel.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.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.

Warning:

**Do NOT instantiate directly.** Instead access the instance via the kernel on [async_kernel.kernel.Kernel.subshell_manager][].

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

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

    Warning:

        **Do NOT instantiate directly.** Instead access the instance via the kernel on [async_kernel.kernel.Kernel.subshell_manager][].
    """

    __slots__ = ["__weakref__"]

    main_shell: Fixed[Self, AsyncInteractiveShell] = Fixed(lambda _: utils.get_kernel().main_shell)
    _main_shell_pending_manager_id: Fixed[Self, str] = Fixed(lambda c: c["owner"].main_shell.pending_manager.id)
    subshells: dict[str, AsyncInteractiveSubshell] = {}
    default_subshell_class = AsyncInteractiveSubshell

    def create_subshell(self, *, 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 self.default_subshell_class(protected=protected)

    def list_subshells(self) -> list[str]:
        return list(self.subshells)

    if TYPE_CHECKING:

        @overload
        def get_shell(self, subshell_id: str) -> AsyncInteractiveSubshell: ...
        @overload
        def get_shell(self, subshell_id: None = ...) -> AsyncInteractiveShell: ...

    def get_shell(
        self,
        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.
        """
        if subshell_id is NoValue:
            subshell_id = ShellPendingManager.active_id()
        if subshell_id is None or subshell_id == self._main_shell_pending_manager_id:
            return self.main_shell
        return self.subshells[subshell_id]

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

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

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

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

create_subshell ¶

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

def create_subshell(self, *, 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 self.default_subshell_class(protected=protected)

get_shell ¶

get_shell(subshell_id: str) -> AsyncInteractiveSubshell

get_shell(subshell_id: None = ...) -> AsyncInteractiveShell

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

def get_shell(
    self,
    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.
    """
    if subshell_id is NoValue:
        subshell_id = ShellPendingManager.active_id()
    if subshell_id is None or subshell_id == self._main_shell_pending_manager_id:
        return self.main_shell
    return self.subshells[subshell_id]

delete_subshell ¶

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

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

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

stop_all_subshells ¶

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

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

    Args:
        force: Passed to [async_kernel.asyncshell.AsyncInteractiveSubshell.stop][].
    """
    for subshell in set(self.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.

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 = self.shell.kernel.subshell_manager.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}")

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 = self.shell.kernel.subshell_manager.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}")

asyncshell

all `module-attribute` ¶

AsyncInteractiveShell ¶

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

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

init_hooks ¶

init_magics ¶

enable_matplotlib ¶

`gui` ¶

context ¶

AsyncInteractiveSubshell ¶

stop ¶

KernelInterruptError ¶

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 ¶

`protected` ¶

get_shell ¶

`subshell_id` ¶

delete_subshell ¶

`subshell_id` ¶

stop_all_subshells ¶

`force` ¶

KernelMagics ¶

connect_info ¶

callers ¶

subshell ¶

asyncshell

__all__ module-attribute ¶

AsyncInteractiveShell ¶

timeout class-attribute instance-attribute ¶

stop_on_error_time_offset class-attribute instance-attribute ¶

init_hooks ¶

init_magics ¶

enable_matplotlib ¶

gui ¶

context ¶

AsyncInteractiveSubshell ¶

stop ¶

KernelInterruptError ¶

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 ¶

protected ¶

get_shell ¶

subshell_id ¶

delete_subshell ¶

subshell_id ¶

stop_all_subshells ¶

force ¶

KernelMagics ¶

connect_info ¶

callers ¶

subshell ¶

all `module-attribute` ¶

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

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

`gui` ¶

`data` ¶

`metadata` ¶

`transient` ¶

`update` ¶

`wait` ¶

`protected` ¶

`subshell_id` ¶

`subshell_id` ¶

`force` ¶