kernel

Classes:

KernelInterrupt –

Raised to interrupt the kernel.
Kernel –

The class containing the handler methods to implement a Jupyter Kernel.

KernelInterrupt ¶

Bases: InterruptedError

Raised to interrupt the kernel.

Source code in src/async_kernel/common.py

class KernelInterrupt(InterruptedError):
    "Raised to interrupt the kernel."

Kernel ¶

Bases: HasInterface[T_interface_co], LoggingConfigurable, Generic[T_interface_co, T_shell_co]

The class containing the handler methods to implement a Jupyter Kernel.

Referenced by:

Reference Reference Highlights

Returned by:

Reference
- interface
  - BaseInterface
    - kernel
    - kernel_class
  - BaseInterface
    - kernel
    - kernel_class
- shell BaseShell kernel
- utils get_kernel

Methods:

running –

The kernel runs in this context.
do_interrupt –

Interrupt/cancel non-silent active execute requests.
apply_patches –

Apply patches returning a callable to reverse the patches.
displayhook –

The global patch for sys.displayhook responsible for python display callbacks.
message_handler –

Schedule handling of the job (msg) with a handler running in a Task managed by a Caller.
get_shell –

Get a shell by subshell_id.
create_subshell –

Create a subshell.
kernel_info_request –

Handle an kernel info request.
comm_info_request –

Handle an comm info request.
execute_request –

Handle an execute request.
complete_request –

Handle an completion request.
is_complete_request –

Handle an is_complete request.
inspect_request –

Handle an inspect request.
history_request –

Handle an history request.
comm_open –

Handle an comm open request.
comm_msg –

Handle an comm msg request.
comm_close –

Handle an comm close request.
interrupt_request –

Handle an interrupt request.
shutdown_request –

Handle an shutdown request.
debug_request –

Handle an debug request.
create_subshell_request –

Handle an create subshell request.
delete_subshell_request –

Handle an delete subshell request.
list_subshell_request –

Handle an list subshell request.
get_parent –

A convenience method to access the 'message' in the current context if there is one.
do_complete –

Matches signature of ipykernel.kernelbase.Kernel.do_complete.
do_inspect –

Matches signature of ipykernel.kernelbase.Kernel.do_inspect.
do_history –

Matches signature of ipykernel.kernelbase.Kernel.do_history.
do_execute –

Matches signature of ipykernel.kernelbase.Kernel.do_execute.
getpass –

Matches signature of ipykernel.kernelbase.Kernel.getpass.
raw_input –

Matches signature of ipykernel.kernelbase.Kernel.raw_input.

Attributes:

help_links –

A list of links provided kernel info request.
supported_features –

A list of features supported by the kernel.
handle_in_shell_thread –

A list of MsgType that are always handled in the shell's thread (typically the MainThread).
handle_in_thread –

A mapping of MsgType to the name of a separate caller (thread) in which to run the handler.
callers (Fixed[Self, dict]) –

A shortcut to the callers dict on the parent.
caller (Fixed[Self, Caller]) –

The caller for the shell thread.
debugger –

The debugger for handling debug requests.
comm_manager –

Creates async_kernel.comm.Comm instances and maintains a mapping to comm_id to Comm instances.
active_execute_requests (Fixed[Self, set[Pending[Any]]]) –

A set of active execute requests that gets updated by the shell.
kernel_info (dict[str, Any]) –

Info provided to a kernel info request.
shell (T_shell_co) –

The shell given the current context.

Source code in src/async_kernel/kernel.py

class Kernel(HasInterface[T_interface_co], LoggingConfigurable, Generic[T_interface_co, T_shell_co]):
    """
    The class containing the handler methods to implement a Jupyter Kernel.
    """

    help_links = traitlets.List(trait=traitlets.Dict()).tag(config=True)
    "A list of links provided kernel info request."

    supported_features = traitlets.List(traitlets.Unicode()).tag(config=True)
    "A list of features supported by the kernel."

    handle_in_shell_thread = traitlets.List(
        traitlets.UseEnum(MsgType),
        [MsgType.comm_msg, MsgType.comm_open, MsgType.comm_close],
    ).tag(config=True)
    """
    A list of `MsgType` that are always handled in the shell's thread (typically the _MainThread_).
    """

    handle_in_thread = traitlets.Dict(key_trait=traitlets.UseEnum(MsgType), value_trait=traitlets.Unicode())
    """
    A mapping of `MsgType` to the name of a separate caller (thread) in which to run the handler.
    """

    callers: Fixed[Self, dict] = Fixed(lambda c: c["owner"].parent.callers)
    "A shortcut to the callers dict on the parent."

    caller: Fixed[Self, Caller] = Fixed(lambda c: c["owner"].callers[Channel.shell])
    "The caller for the shell thread."

    debugger = Fixed(Debugger)
    "The debugger for handling debug requests."

    comm_manager = Fixed(CommManager)
    "Creates [async_kernel.comm.Comm][] instances and maintains a mapping to `comm_id` to `Comm` instances."

    active_execute_requests: Fixed[Self, set[Pending[Any]]] = Fixed(set)
    "A set of active execute requests that gets updated by the shell."

    _interrupt_message = "Kernel interrupted"

    _restart = False
    _handler_cache: ClassVar[dict[tuple[str | None, MsgType, Callable], HandlerType]] = {}
    _subshells: dict[str, T_shell_co]
    _interrupt_requested: None | Pending = None

    @traitlets.default("help_links")
    def _default_help_links(self) -> tuple[dict[str, str], ...]:
        return (
            {
                "text": "Async Kernel Reference ",
                "url": "https://fleming79.github.io/async-kernel/",
            },
            {
                "text": "IPython Reference",
                "url": "https://ipython.readthedocs.io/en/stable/",
            },
            {
                "text": "IPython magic Reference",
                "url": "https://ipython.readthedocs.io/en/stable/interactive/magics.html",
            },
            {
                "text": "Matplotlib ipympl Reference",
                "url": "https://matplotlib.org/ipympl/",
            },
            {
                "text": "Matplotlib Reference",
                "url": "https://matplotlib.org/contents.html",
            },
        )

    @traitlets.default("handle_in_thread")
    def _default_handle_in_thread(self) -> dict[MsgType, str]:
        return {
            MsgType.inspect_request: "language_server",
            MsgType.complete_request: "language_server",
            MsgType.is_complete_request: "language_server",
        }

    @traitlets.default("supported_features")
    def _default_supported_features(self) -> list[str]:
        features = ["kernel subshells"]
        if self.debugger.enabled:
            features.append("debugger")
        return features

    @property
    def kernel_info(self) -> dict[str, Any]:
        "Info provided to a kernel info request."
        return {
            "protocol_version": async_kernel.kernel_protocol_version,
            "implementation": async_kernel.distribution_name,
            "implementation_version": async_kernel.__version__,
            "language_info": async_kernel.kernel_protocol_version_info,
            "banner": self.shell.banner,
            "help_links": self.help_links,
            "debugger": self.debugger.enabled,
            "supported_features": self.supported_features,
        }

    @property
    def main_shell(self) -> T_shell_co:
        try:
            return self._main_shell
        except AttributeError:
            msg = "The main_shell is only available once the kernel is started."
            raise RuntimeError(msg) from None

    @property
    def shell(self) -> T_shell_co:
        """
        The shell given the current context.

        Notes:
            - The `subshell_id` of the main shell is `None`.
        """
        return self.get_shell()

    @property
    def subshells(self) -> dict[str, T_shell_co]:
        return self._subshells.copy()

    def __init__(self, parent: T_interface_co, shell_class: type[T_shell_co]) -> None:
        self._subshells = {}
        assert self.parent is parent
        super().__init__()
        self._shell_class = shell_class

    @asynccontextmanager
    async def running(self) -> AsyncGenerator[None]:
        """
        The kernel runs in this context.

        Notes:
            - Entered by the interface (parent).
            - Overload as required.
        """
        remove_patch = self.apply_patches()
        try:
            self._main_shell = self._shell_class(protected=True, is_mainshell=True)
            async with self._main_shell.mainshell_running():
                self.log.info("Kernel started")
                yield
        finally:
            self.log.info("Kernel stopped")
            for subshell in self._subshells.copy().values():
                subshell.stop(force=True)
            for comm in self.comm_manager.comms.copy().values():
                comm.close(deleting=True)
            remove_patch()
            self._handler_cache.clear()

    def _signal_handler(self, signum, frame: FrameType | None) -> None:
        "Handle interrupt signals."

        if pen := self._interrupt_requested:
            self._interrupt_requested = None
            if not pen.done():
                with enable_signal_safety():
                    pen.set_result(None)
                raise KernelInterrupt
        else:
            signal.default_int_handler(signum, frame)

    async def do_interrupt(self) -> None:
        """
        Interrupt/cancel non-silent active execute requests.
        """
        assert Caller() is self.callers[Channel.control], "Must be called from the control thread."
        for pen in self.active_execute_requests.copy():
            if not pen.metadata.get("kwargs", {}).get("silent", False):
                pen.cancel(self._interrupt_message)
        if (
            (sys.platform != "emscripten")
            and (not self.debugger.enabled or not self.debugger.stopped_threads)
            and (caller := Caller("MainThread")).running  # Can only interrupt the main thread
        ):
            self._interrupt_requested = pen = Pending()
            caller.call_direct(lambda: pen.set_result(None))
            try:
                await pen.wait(result=False, timeout=1, protect=True)
            except TimeoutError:
                if sys.platform == "win32":
                    signal.raise_signal(signal.SIGINT)
                else:
                    os.kill(os.getpid(), signal.SIGINT)
                await pen.wait(result=False, timeout=10, protect=True)

    def _patch_signal(self) -> Callable[[], None]:

        with contextlib.suppress(ValueError, AttributeError):
            import signal  # noqa: PLC0415

            sig = signal.signal(signal.SIGINT, self._signal_handler)

            def restore() -> None:
                signal.signal(signal.SIGINT, sig)

            return restore
        return lambda: None

    def apply_patches(self) -> Callable[[], None]:
        """Apply patches returning a callable to reverse the patches."""

        original = sys.displayhook, builtins.input, getpass.getpass
        builtins.input, sys.displayhook, getpass.getpass = self.raw_input, self.displayhook, self.getpass
        restore_comm = self.comm_manager.patch_comm()
        restore_stdout = OutStream("stdout").patch()
        restore_stderr = OutStream("stderr").patch()
        restore_signal = self._patch_signal()

        def restore() -> None:
            sys.displayhook, builtins.input, getpass.getpass = original
            restore_stdout()
            restore_stderr()
            restore_comm()
            restore_signal()

        return restore

    def displayhook(self, result: Any):
        "The global patch for [sys.displayhook][] responsible for python display callbacks."
        self.get_shell().displayhook(result)

    def _shell_created(self, shell: T_shell_co) -> None:  # pyright: ignore[reportGeneralTypeIssues]
        "Called by `BaseShell.__init__`"
        if shell.subshell_id:
            self._subshells[shell.subshell_id] = shell

    def _subshell_stopped(self, shell: T_shell_co) -> None:  # pyright: ignore[reportGeneralTypeIssues]
        "Called by `BaseShell.stop`"
        if subshell_id := shell.subshell_id:
            self._subshells.pop(subshell_id, None)
        for key in list(self._handler_cache):
            if key[0] == subshell_id:
                self._handler_cache.pop(key, None)

    def _get_handler(
        self,
        job: Job,
        send_reply: Callable[[Job, dict], CoroutineType[Any, Any, None]],
        iopub_send: Callable,
    ) -> HandlerType:
        try:
            subshell_id = job["msg"]["content"]["subshell_id"]
        except KeyError:
            try:
                subshell_id = job["msg"]["header"]["subshell_id"]  # pyright: ignore[reportTypedDictNotRequiredAccess]
            except KeyError:
                subshell_id = None
        msg_type = MsgType(job["msg"]["header"]["msg_type"])

        if msg_type is MsgType.execute_request:
            key = (subshell_id, msg_type, send_reply)
        else:
            key = (None, msg_type, send_reply)
        try:
            return self._handler_cache[key]
        except KeyError:
            handler: HandlerType = getattr(self, msg_type)

            @functools.wraps(handler)
            async def run_handler(job: Job) -> None:
                job_token = utils._job_var.set(job)  # pyright: ignore[reportPrivateUsage]
                subshell_token = ShellPendingManager._id_contextvar.set(subshell_id)  # pyright: ignore[reportPrivateUsage]

                try:
                    iopub_send(
                        msg_or_type="status",
                        parent=job["msg"],
                        content={"execution_state": "busy"},
                        ident=b"kernel.status",
                    )
                    if (content := await handler(job)) is not None:
                        await send_reply(job, content)
                except Exception as e:
                    await send_reply(job, utils.error_to_content(e))
                    self.log.exception("Exception in message handler:", exc_info=e)
                finally:
                    utils._job_var.reset(job_token)  # pyright: ignore[reportPrivateUsage]
                    ShellPendingManager._id_contextvar.reset(subshell_token)  # pyright: ignore[reportPrivateUsage]
                    iopub_send(
                        msg_or_type="status",
                        parent=job["msg"],
                        content={"execution_state": "idle"},
                        ident=b"kernel.status",
                    )
                    del job

            self._handler_cache[key] = run_handler
            return run_handler

    def message_handler(
        self,
        job: Job,
        send_reply: Callable[[Job, dict], CoroutineType[Any, Any, None]],
        iopub_send: Callable,
        /,
    ) -> None:
        """
        Schedule handling of the job (msg) with a handler running in a Task managed by a Caller.

        Each `msg_type` runs in a separate task, possibly in a separate thread and event loop.
        Typically, jobs are queued for execution by either the 'shell' or 'control' caller using
        [queue_call][async_kernel.caller.Caller.queue_call].

        'execute_request' messages can also specify alternate run modes:
            - task: Run the execute request as a task.
            - thread: Run the execute request in a worker thread.

            The alternate run mode can be specified in a few ways:
            - as a comment on the first line of the code block `# task` or `# thread`.
            - As a tag `thread` or `task`

        Args:
            job: A dict with the msg and supporting details.
            send_reply: The function for the handler to use to send the reply to the message.
        """
        handler = self._get_handler(job, send_reply, iopub_send)

        run_mode: RunMode | CallerCreateOptions | None = None
        msg_type = MsgType(job["msg"]["header"]["msg_type"])

        if msg_type is MsgType.execute_request:
            caller = self.callers[job["msg"]["channel"]]
            try:
                run_mode = next(mode for tag in utils.get_tags(job) if (mode := RunMode.to_runmode(tag)))
            except StopIteration:
                if content := job["msg"].get("content", {}):
                    if (code := content.get("code")) and (
                        mode := RunMode.to_runmode(code.strip().split("\n", maxsplit=1)[0])
                    ):
                        run_mode = mode
                    if content.get("silent"):
                        run_mode = RunMode.task

        elif msg_type in self.handle_in_shell_thread:
            caller = self.callers[Channel.shell]
        else:
            caller = self.callers[Channel.control]
            if thread_name := self.handle_in_thread.get(msg_type):
                caller = caller.get(name=thread_name, no_debug=True)

        match run_mode:
            case RunMode.queue | None:
                caller.queue_call(handler, job)
            case RunMode.task:
                caller.call_soon(handler, job)
            case RunMode.thread:
                caller.to_thread(handler, job)
            case _ as options:
                caller.get(**options).call_soon(handler, job)

        self.log.debug("***handle message %s*** %s %s %s", msg_type, run_mode, handler, job)

    def get_shell(self, subshell_id: str | None | NoValue = NoValue) -> T_shell_co:  # pyright: ignore[reportInvalidTypeForm]
        """
        Get a shell by `subshell_id`.

        Args:
            subshell_id: The id of an existing subshell.
        """
        if (subshell_id := subshell_id if subshell_id is not NoValue else ShellPendingManager.active_id()) is None:
            return self.main_shell
        return self._subshells.get(subshell_id) or self.main_shell

    def create_subshell(self, *, protected: bool = False) -> T_shell_co:
        """
        Create a subshell.

        Use [`shell.stop(force=True)`][async_kernel.shell.base.BaseShell.stop] to stop a
        protected subshell when it is no longer required.

        Args:
            protected: Protect the subshell from accidental deletion.
        Tip:
            - `await shell.ready` to ensure the shell is 'ready'.
        """

        return self._shell_class(protected=protected)

    async def kernel_info_request(self, job: Job[Content], /) -> Content:
        """Handle an [kernel info request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#kernel-info)."""
        return self.kernel_info

    async def comm_info_request(self, job: Job[Content], /) -> Content:
        """Handle an [comm info request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#comm-info)."""
        c = job["msg"]["content"]
        target_name = c.get("target_name", None)
        comms = {
            k: {"target_name": v.target_name}
            for (k, v) in self.comm_manager.comms.copy().items()
            if v.target_name == target_name or target_name is None
        }
        return {"comms": comms}

    async def execute_request(self, job: Job[ExecuteContent], /) -> Content:
        """Handle an [execute request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#execute)."""
        return await self.shell.do_execute(
            cell_id=job["msg"]["metadata"].get("cellId"),
            received_time=job["received_time"],
            tags=job["msg"]["metadata"].get("tags", ()),
            **job["msg"]["content"],  # pyright: ignore[reportArgumentType]
        )

    async def complete_request(self, job: Job[Content], /) -> Content:
        """Handle an [completion request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#completion)."""
        return await self.shell.do_complete(
            code=job["msg"]["content"].get("code", ""), cursor_pos=job["msg"]["content"].get("cursor_pos", 0)
        )

    async def is_complete_request(self, job: Job[Content], /) -> Content:
        """Handle an [is_complete request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#code-completeness)."""
        return await self.shell.is_complete(job["msg"]["content"].get("code", ""))

    async def inspect_request(self, job: Job[Content], /) -> Content:
        """Handle an [inspect request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#introspection)."""
        c = job["msg"]["content"]
        return await self.shell.do_inspect(
            code=c.get("code", ""),
            cursor_pos=c.get("cursor_pos", 0),
            detail_level=c.get("detail_level", 0),
        )

    async def history_request(self, job: Job[Content], /) -> Content:
        """Handle an [history request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#history)."""
        return await self.shell.do_history(**job["msg"]["content"])

    async def comm_open(self, job: Job[Content], /) -> None:
        """Handle an [comm open request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#opening-a-comm)."""
        self.comm_manager.comm_open(stream=None, ident=None, msg=job["msg"])  # pyright: ignore[reportArgumentType]

    async def comm_msg(self, job: Job[Content], /) -> None:
        """Handle an [comm msg request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#comm-messages)."""
        self.comm_manager.comm_msg(stream=None, ident=None, msg=job["msg"])  # pyright: ignore[reportArgumentType]

    async def comm_close(self, job: Job[Content], /) -> None:
        """Handle an [comm close request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#tearing-down-comms)."""
        self.comm_manager.comm_close(stream=None, ident=None, msg=job["msg"])  # pyright: ignore[reportArgumentType]

    async def interrupt_request(self, job: Job[Content], /) -> Content:
        """Handle an [interrupt request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#kernel-interrupt)."""
        await self.do_interrupt()
        return {}

    async def shutdown_request(self, job: Job[Content], /) -> Content:
        """Handle an [shutdown request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#kernel-shutdown)."""
        self._restart = job["msg"]["content"].get("restart", False)
        self.parent.stop()
        return {"restart": self._restart}

    async def debug_request(self, job: Job[Content], /) -> Content:
        """Handle an [debug request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#debug-request)."""
        return await self.debugger.process_request(job["msg"]["content"])

    async def create_subshell_request(self: Kernel, job: Job[Content], /) -> Content:
        """Handle an [create subshell request](https://jupyter.org/enhancement-proposals/91-kernel-subshells/kernel-subshells.html#create-subshell)."""
        async with self.caller.create_pending_group():
            shell = self.create_subshell(protected=False)
            return {"subshell_id": shell.subshell_id}

    async def delete_subshell_request(self, job: Job[Content], /) -> Content:
        """Handle an [delete subshell request](https://jupyter.org/enhancement-proposals/91-kernel-subshells/kernel-subshells.html#delete-subshell)."""
        if (subshell_id := job["msg"]["content"]["subshell_id"]) and (subshell := self._subshells.get(subshell_id)):
            subshell.stop()
        return {}

    async def list_subshell_request(self, job: Job[Content], /) -> Content:
        """Handle an [list subshell request](https://jupyter.org/enhancement-proposals/91-kernel-subshells/kernel-subshells.html#list-subshells)."""
        return {"subshell_id": list(self._subshells)}

    def get_parent(self) -> Message[dict[str, Any]] | None:
        """
        A convenience method to access the 'message' in the current context if there is one.

        'parent' is the parameter name used by [Session.send][jupyter_client.session.Session.send] to provide context when sending a reply.

        See also:
            - [ipywidgets.Output][ipywidgets.widgets.widget_output.Output]:
                Uses `get_ipython().kernel.get_parent()` to obtain the `msg_id` which
                is used to 'capture' output when its context has been acquired.
        """
        return utils.get_parent_message()

    async def do_complete(self, code: str, cursor_pos: int | None) -> Content:
        "Matches signature of [ipykernel.kernelbase.Kernel.do_complete][]."
        return await self.shell.do_complete(code=code, cursor_pos=cursor_pos)

    async def do_inspect(
        self, code: str, cursor_pos: int = 0, detail_level: Literal[0, 1] = 0, omit_sections=()
    ) -> Content:
        "Matches signature of [ipykernel.kernelbase.Kernel.do_inspect][]."
        return await self.shell.do_inspect(code=code, cursor_pos=cursor_pos, detail_level=detail_level)

    async def do_history(
        self,
        hist_access_type,
        output,
        raw,
        session=None,
        start=None,
        stop=None,
        n=None,
        pattern=None,
        unique=False,
    ) -> Content:
        "Matches signature of [ipykernel.kernelbase.Kernel.do_history][]."
        return await self.shell.do_history(
            output=output,
            raw=raw,
            hist_access_type=hist_access_type,
            session=session,  # pyright: ignore[reportArgumentType]
            start=start,  # pyright: ignore[reportArgumentType]
            stop=stop,
        )

    async def do_execute(
        self,
        code: str,
        silent: bool = True,
        store_history: bool = True,
        user_expressions: dict[str, str] | None = None,
        allow_stdin: bool = False,
        *,
        cell_meta: dict[str, Any] | None = None,
        cell_id: str | None = None,
        **_ignored,
    ) -> Content:
        "Matches signature of [ipykernel.kernelbase.Kernel.do_execute][]."
        return await self.shell.do_execute(
            code=code,
            silent=silent,
            store_history=store_history,
            user_expressions=user_expressions,
            allow_stdin=allow_stdin,
            cell_id=cell_id,
        )

    def getpass(self, prompt="", stream=None) -> str:
        "Matches signature of [ipykernel.kernelbase.Kernel.getpass][]."
        return self.parent.input_request(str(prompt), password=True)

    def raw_input(self, prompt="") -> str:
        "Matches signature of [ipykernel.kernelbase.Kernel.raw_input][]."
        return self.parent.input_request(str(prompt), password=False)

help_links `class-attribute` `instance-attribute` ¶

help_links = traitlets.List(trait=traitlets.Dict()).tag(config=True)

A list of links provided kernel info request.

supported_features `class-attribute` `instance-attribute` ¶

supported_features = traitlets.List(traitlets.Unicode()).tag(config=True)

A list of features supported by the kernel.

handle_in_shell_thread `class-attribute` `instance-attribute` ¶

handle_in_shell_thread = traitlets.List(
    traitlets.UseEnum(MsgType), [MsgType.comm_msg, MsgType.comm_open, MsgType.comm_close]
).tag(config=True)

A list of MsgType that are always handled in the shell's thread (typically the MainThread).

handle_in_thread `class-attribute` `instance-attribute` ¶

handle_in_thread = traitlets.Dict(
    key_trait=traitlets.UseEnum(MsgType), value_trait=traitlets.Unicode()
)

A mapping of MsgType to the name of a separate caller (thread) in which to run the handler.

callers `class-attribute` `instance-attribute` ¶

callers: Fixed[Self, dict] = Fixed(lambda c: c['owner'].parent.callers)

A shortcut to the callers dict on the parent.

caller `class-attribute` `instance-attribute` ¶

caller: Fixed[Self, Caller] = Fixed(lambda c: c['owner'].callers[Channel.shell])

The caller for the shell thread.

debugger `class-attribute` `instance-attribute` ¶

debugger = Fixed(Debugger)

The debugger for handling debug requests.

comm_manager `class-attribute` `instance-attribute` ¶

comm_manager = Fixed(CommManager)

Creates async_kernel.comm.Comm instances and maintains a mapping to comm_id to Comm instances.

active_execute_requests `class-attribute` `instance-attribute` ¶

active_execute_requests: Fixed[Self, set[Pending[Any]]] = Fixed(set)

A set of active execute requests that gets updated by the shell.

kernel_info `property` ¶

kernel_info: dict[str, Any]

Info provided to a kernel info request.

shell `property` ¶

shell: T_shell_co

The shell given the current context.

Notes

The subshell_id of the main shell is None.

running `async` ¶

running() -> AsyncGenerator[None]

The kernel runs in this context.

Notes

Entered by the interface (parent).
Overload as required.

Source code in src/async_kernel/kernel.py

@asynccontextmanager
async def running(self) -> AsyncGenerator[None]:
    """
    The kernel runs in this context.

    Notes:
        - Entered by the interface (parent).
        - Overload as required.
    """
    remove_patch = self.apply_patches()
    try:
        self._main_shell = self._shell_class(protected=True, is_mainshell=True)
        async with self._main_shell.mainshell_running():
            self.log.info("Kernel started")
            yield
    finally:
        self.log.info("Kernel stopped")
        for subshell in self._subshells.copy().values():
            subshell.stop(force=True)
        for comm in self.comm_manager.comms.copy().values():
            comm.close(deleting=True)
        remove_patch()
        self._handler_cache.clear()

do_interrupt `async` ¶

do_interrupt() -> None

Interrupt/cancel non-silent active execute requests.

Source code in src/async_kernel/kernel.py

async def do_interrupt(self) -> None:
    """
    Interrupt/cancel non-silent active execute requests.
    """
    assert Caller() is self.callers[Channel.control], "Must be called from the control thread."
    for pen in self.active_execute_requests.copy():
        if not pen.metadata.get("kwargs", {}).get("silent", False):
            pen.cancel(self._interrupt_message)
    if (
        (sys.platform != "emscripten")
        and (not self.debugger.enabled or not self.debugger.stopped_threads)
        and (caller := Caller("MainThread")).running  # Can only interrupt the main thread
    ):
        self._interrupt_requested = pen = Pending()
        caller.call_direct(lambda: pen.set_result(None))
        try:
            await pen.wait(result=False, timeout=1, protect=True)
        except TimeoutError:
            if sys.platform == "win32":
                signal.raise_signal(signal.SIGINT)
            else:
                os.kill(os.getpid(), signal.SIGINT)
            await pen.wait(result=False, timeout=10, protect=True)

apply_patches ¶

apply_patches() -> Callable[[], None]

Apply patches returning a callable to reverse the patches.

Source code in src/async_kernel/kernel.py

def apply_patches(self) -> Callable[[], None]:
    """Apply patches returning a callable to reverse the patches."""

    original = sys.displayhook, builtins.input, getpass.getpass
    builtins.input, sys.displayhook, getpass.getpass = self.raw_input, self.displayhook, self.getpass
    restore_comm = self.comm_manager.patch_comm()
    restore_stdout = OutStream("stdout").patch()
    restore_stderr = OutStream("stderr").patch()
    restore_signal = self._patch_signal()

    def restore() -> None:
        sys.displayhook, builtins.input, getpass.getpass = original
        restore_stdout()
        restore_stderr()
        restore_comm()
        restore_signal()

    return restore

displayhook ¶

displayhook(result: Any)

The global patch for sys.displayhook responsible for python display callbacks.

Referenced by:

Reference ipshell IPShell displayhook

Source code in src/async_kernel/kernel.py

def displayhook(self, result: Any):
    "The global patch for [sys.displayhook][] responsible for python display callbacks."
    self.get_shell().displayhook(result)

message_handler ¶

message_handler(
    job: Job,
    send_reply: Callable[[Job, dict], CoroutineType[Any, Any, None]],
    iopub_send: Callable,
) -> None

Schedule handling of the job (msg) with a handler running in a Task managed by a Caller.

Each msg_type runs in a separate task, possibly in a separate thread and event loop. Typically, jobs are queued for execution by either the 'shell' or 'control' caller using queue_call.

'execute_request' messages can also specify alternate run modes: - task: Run the execute request as a task. - thread: Run the execute request in a worker thread.

The alternate run mode can be specified in a few ways:
- as a comment on the first line of the code block `# task` or `# thread`.
- As a tag `thread` or `task`

Parameters:

job ¶
(Job) –

A dict with the msg and supporting details.
send_reply ¶
(Callable[[Job, dict], CoroutineType[Any, Any, None]]) –

The function for the handler to use to send the reply to the message.

Source code in src/async_kernel/kernel.py

def message_handler(
    self,
    job: Job,
    send_reply: Callable[[Job, dict], CoroutineType[Any, Any, None]],
    iopub_send: Callable,
    /,
) -> None:
    """
    Schedule handling of the job (msg) with a handler running in a Task managed by a Caller.

    Each `msg_type` runs in a separate task, possibly in a separate thread and event loop.
    Typically, jobs are queued for execution by either the 'shell' or 'control' caller using
    [queue_call][async_kernel.caller.Caller.queue_call].

    'execute_request' messages can also specify alternate run modes:
        - task: Run the execute request as a task.
        - thread: Run the execute request in a worker thread.

        The alternate run mode can be specified in a few ways:
        - as a comment on the first line of the code block `# task` or `# thread`.
        - As a tag `thread` or `task`

    Args:
        job: A dict with the msg and supporting details.
        send_reply: The function for the handler to use to send the reply to the message.
    """
    handler = self._get_handler(job, send_reply, iopub_send)

    run_mode: RunMode | CallerCreateOptions | None = None
    msg_type = MsgType(job["msg"]["header"]["msg_type"])

    if msg_type is MsgType.execute_request:
        caller = self.callers[job["msg"]["channel"]]
        try:
            run_mode = next(mode for tag in utils.get_tags(job) if (mode := RunMode.to_runmode(tag)))
        except StopIteration:
            if content := job["msg"].get("content", {}):
                if (code := content.get("code")) and (
                    mode := RunMode.to_runmode(code.strip().split("\n", maxsplit=1)[0])
                ):
                    run_mode = mode
                if content.get("silent"):
                    run_mode = RunMode.task

    elif msg_type in self.handle_in_shell_thread:
        caller = self.callers[Channel.shell]
    else:
        caller = self.callers[Channel.control]
        if thread_name := self.handle_in_thread.get(msg_type):
            caller = caller.get(name=thread_name, no_debug=True)

    match run_mode:
        case RunMode.queue | None:
            caller.queue_call(handler, job)
        case RunMode.task:
            caller.call_soon(handler, job)
        case RunMode.thread:
            caller.to_thread(handler, job)
        case _ as options:
            caller.get(**options).call_soon(handler, job)

    self.log.debug("***handle message %s*** %s %s %s", msg_type, run_mode, handler, job)

get_shell ¶

get_shell(subshell_id: str | None | NoValue = NoValue) -> T_shell_co

Get a shell by subshell_id.

Parameters:

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

The id of an existing subshell.

Source code in src/async_kernel/kernel.py

def get_shell(self, subshell_id: str | None | NoValue = NoValue) -> T_shell_co:  # pyright: ignore[reportInvalidTypeForm]
    """
    Get a shell by `subshell_id`.

    Args:
        subshell_id: The id of an existing subshell.
    """
    if (subshell_id := subshell_id if subshell_id is not NoValue else ShellPendingManager.active_id()) is None:
        return self.main_shell
    return self._subshells.get(subshell_id) or self.main_shell

create_subshell ¶

create_subshell(*, protected: bool = False) -> T_shell_co

Create a subshell.

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

Parameters:

protected ¶
(bool, default: False ) –

Protect the subshell from accidental deletion.

Tip: - await shell.ready to ensure the shell is 'ready'.

Source code in src/async_kernel/kernel.py

def create_subshell(self, *, protected: bool = False) -> T_shell_co:
    """
    Create a subshell.

    Use [`shell.stop(force=True)`][async_kernel.shell.base.BaseShell.stop] to stop a
    protected subshell when it is no longer required.

    Args:
        protected: Protect the subshell from accidental deletion.
    Tip:
        - `await shell.ready` to ensure the shell is 'ready'.
    """

    return self._shell_class(protected=protected)

kernel_info_request `async` ¶

kernel_info_request(job: Job[Content]) -> Content

Handle an kernel info request.

Referenced by:

Reference typing MsgType

Source code in src/async_kernel/kernel.py

async def kernel_info_request(self, job: Job[Content], /) -> Content:
    """Handle an [kernel info request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#kernel-info)."""
    return self.kernel_info

comm_info_request `async` ¶

comm_info_request(job: Job[Content]) -> Content

Handle an comm info request.

Referenced by:

Reference typing MsgType comm_info_request

Source code in src/async_kernel/kernel.py

async def comm_info_request(self, job: Job[Content], /) -> Content:
    """Handle an [comm info request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#comm-info)."""
    c = job["msg"]["content"]
    target_name = c.get("target_name", None)
    comms = {
        k: {"target_name": v.target_name}
        for (k, v) in self.comm_manager.comms.copy().items()
        if v.target_name == target_name or target_name is None
    }
    return {"comms": comms}

execute_request `async` ¶

execute_request(job: Job[ExecuteContent]) -> Content

Handle an execute request.

Referenced by:

Reference typing MsgType execute_request

Source code in src/async_kernel/kernel.py

async def execute_request(self, job: Job[ExecuteContent], /) -> Content:
    """Handle an [execute request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#execute)."""
    return await self.shell.do_execute(
        cell_id=job["msg"]["metadata"].get("cellId"),
        received_time=job["received_time"],
        tags=job["msg"]["metadata"].get("tags", ()),
        **job["msg"]["content"],  # pyright: ignore[reportArgumentType]
    )

complete_request `async` ¶

complete_request(job: Job[Content]) -> Content

Handle an completion request.

Referenced by:

Reference typing MsgType complete_request

Source code in src/async_kernel/kernel.py

async def complete_request(self, job: Job[Content], /) -> Content:
    """Handle an [completion request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#completion)."""
    return await self.shell.do_complete(
        code=job["msg"]["content"].get("code", ""), cursor_pos=job["msg"]["content"].get("cursor_pos", 0)
    )

is_complete_request `async` ¶

is_complete_request(job: Job[Content]) -> Content

Handle an is_complete request.

Referenced by:

Reference typing MsgType

Source code in src/async_kernel/kernel.py

async def is_complete_request(self, job: Job[Content], /) -> Content:
    """Handle an [is_complete request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#code-completeness)."""
    return await self.shell.is_complete(job["msg"]["content"].get("code", ""))

inspect_request `async` ¶

inspect_request(job: Job[Content]) -> Content

Handle an inspect request.

Referenced by:

Reference typing MsgType

Source code in src/async_kernel/kernel.py

async def inspect_request(self, job: Job[Content], /) -> Content:
    """Handle an [inspect request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#introspection)."""
    c = job["msg"]["content"]
    return await self.shell.do_inspect(
        code=c.get("code", ""),
        cursor_pos=c.get("cursor_pos", 0),
        detail_level=c.get("detail_level", 0),
    )

history_request `async` ¶

history_request(job: Job[Content]) -> Content

Handle an history request.

Referenced by:

Reference typing MsgType

Source code in src/async_kernel/kernel.py

async def history_request(self, job: Job[Content], /) -> Content:
    """Handle an [history request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#history)."""
    return await self.shell.do_history(**job["msg"]["content"])

comm_open `async` ¶

comm_open(job: Job[Content]) -> None

Handle an comm open request.

Referenced by:

Reference typing MsgType comm_open

Source code in src/async_kernel/kernel.py

async def comm_open(self, job: Job[Content], /) -> None:
    """Handle an [comm open request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#opening-a-comm)."""
    self.comm_manager.comm_open(stream=None, ident=None, msg=job["msg"])  # pyright: ignore[reportArgumentType]

comm_msg `async` ¶

comm_msg(job: Job[Content]) -> None

Handle an comm msg request.

Referenced by:

Reference typing MsgType

Source code in src/async_kernel/kernel.py

async def comm_msg(self, job: Job[Content], /) -> None:
    """Handle an [comm msg request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#comm-messages)."""
    self.comm_manager.comm_msg(stream=None, ident=None, msg=job["msg"])  # pyright: ignore[reportArgumentType]

comm_close `async` ¶

comm_close(job: Job[Content]) -> None

Handle an comm close request.

Referenced by:

Reference typing MsgType

Source code in src/async_kernel/kernel.py

async def comm_close(self, job: Job[Content], /) -> None:
    """Handle an [comm close request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#tearing-down-comms)."""
    self.comm_manager.comm_close(stream=None, ident=None, msg=job["msg"])  # pyright: ignore[reportArgumentType]

interrupt_request `async` ¶

interrupt_request(job: Job[Content]) -> Content

Handle an interrupt request.

Referenced by:

Reference typing MsgType

Source code in src/async_kernel/kernel.py

async def interrupt_request(self, job: Job[Content], /) -> Content:
    """Handle an [interrupt request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#kernel-interrupt)."""
    await self.do_interrupt()
    return {}

shutdown_request `async` ¶

shutdown_request(job: Job[Content]) -> Content

Handle an shutdown request.

Referenced by:

Reference typing MsgType

Source code in src/async_kernel/kernel.py

async def shutdown_request(self, job: Job[Content], /) -> Content:
    """Handle an [shutdown request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#kernel-shutdown)."""
    self._restart = job["msg"]["content"].get("restart", False)
    self.parent.stop()
    return {"restart": self._restart}

debug_request `async` ¶

debug_request(job: Job[Content]) -> Content

Handle an debug request.

Referenced by:

Reference typing MsgType list_subshell_request

Source code in src/async_kernel/kernel.py

async def debug_request(self, job: Job[Content], /) -> Content:
    """Handle an [debug request](https://jupyter-client.readthedocs.io/en/stable/messaging.html#debug-request)."""
    return await self.debugger.process_request(job["msg"]["content"])

create_subshell_request `async` ¶

create_subshell_request(job: Job[Content]) -> Content

Handle an create subshell request.

Referenced by:

Reference typing MsgType create_subshell_request

Source code in src/async_kernel/kernel.py

async def create_subshell_request(self: Kernel, job: Job[Content], /) -> Content:
    """Handle an [create subshell request](https://jupyter.org/enhancement-proposals/91-kernel-subshells/kernel-subshells.html#create-subshell)."""
    async with self.caller.create_pending_group():
        shell = self.create_subshell(protected=False)
        return {"subshell_id": shell.subshell_id}

delete_subshell_request `async` ¶

delete_subshell_request(job: Job[Content]) -> Content

Handle an delete subshell request.

Referenced by:

Reference typing MsgType

Source code in src/async_kernel/kernel.py

async def delete_subshell_request(self, job: Job[Content], /) -> Content:
    """Handle an [delete subshell request](https://jupyter.org/enhancement-proposals/91-kernel-subshells/kernel-subshells.html#delete-subshell)."""
    if (subshell_id := job["msg"]["content"]["subshell_id"]) and (subshell := self._subshells.get(subshell_id)):
        subshell.stop()
    return {}

list_subshell_request `async` ¶

list_subshell_request(job: Job[Content]) -> Content

Handle an list subshell request.

Source code in src/async_kernel/kernel.py

async def list_subshell_request(self, job: Job[Content], /) -> Content:
    """Handle an [list subshell request](https://jupyter.org/enhancement-proposals/91-kernel-subshells/kernel-subshells.html#list-subshells)."""
    return {"subshell_id": list(self._subshells)}

get_parent ¶

get_parent() -> Message[dict[str, Any]] | None

A convenience method to access the 'message' in the current context if there is one.

'parent' is the parameter name used by Session.send to provide context when sending a reply.

do_complete `async` ¶

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

Matches signature of ipykernel.kernelbase.Kernel.do_complete.

Source code in src/async_kernel/kernel.py

async def do_complete(self, code: str, cursor_pos: int | None) -> Content:
    "Matches signature of [ipykernel.kernelbase.Kernel.do_complete][]."
    return await self.shell.do_complete(code=code, cursor_pos=cursor_pos)

do_inspect `async` ¶

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

Matches signature of ipykernel.kernelbase.Kernel.do_inspect.

Source code in src/async_kernel/kernel.py

async def do_inspect(
    self, code: str, cursor_pos: int = 0, detail_level: Literal[0, 1] = 0, omit_sections=()
) -> Content:
    "Matches signature of [ipykernel.kernelbase.Kernel.do_inspect][]."
    return await self.shell.do_inspect(code=code, cursor_pos=cursor_pos, detail_level=detail_level)

do_history `async` ¶

do_history(
    hist_access_type,
    output,
    raw,
    session=None,
    start=None,
    stop=None,
    n=None,
    pattern=None,
    unique=False,
) -> Content

Matches signature of ipykernel.kernelbase.Kernel.do_history.

Source code in src/async_kernel/kernel.py

async def do_history(
    self,
    hist_access_type,
    output,
    raw,
    session=None,
    start=None,
    stop=None,
    n=None,
    pattern=None,
    unique=False,
) -> Content:
    "Matches signature of [ipykernel.kernelbase.Kernel.do_history][]."
    return await self.shell.do_history(
        output=output,
        raw=raw,
        hist_access_type=hist_access_type,
        session=session,  # pyright: ignore[reportArgumentType]
        start=start,  # pyright: ignore[reportArgumentType]
        stop=stop,
    )

do_execute `async` ¶

do_execute(
    code: str,
    silent: bool = True,
    store_history: bool = True,
    user_expressions: dict[str, str] | None = None,
    allow_stdin: bool = False,
    *,
    cell_meta: dict[str, Any] | None = None,
    cell_id: str | None = None,
    **_ignored,
) -> Content

Matches signature of ipykernel.kernelbase.Kernel.do_execute.

Source code in src/async_kernel/kernel.py

async def do_execute(
    self,
    code: str,
    silent: bool = True,
    store_history: bool = True,
    user_expressions: dict[str, str] | None = None,
    allow_stdin: bool = False,
    *,
    cell_meta: dict[str, Any] | None = None,
    cell_id: str | None = None,
    **_ignored,
) -> Content:
    "Matches signature of [ipykernel.kernelbase.Kernel.do_execute][]."
    return await self.shell.do_execute(
        code=code,
        silent=silent,
        store_history=store_history,
        user_expressions=user_expressions,
        allow_stdin=allow_stdin,
        cell_id=cell_id,
    )

getpass ¶

getpass(prompt='', stream=None) -> str

Matches signature of ipykernel.kernelbase.Kernel.getpass.

Source code in src/async_kernel/kernel.py

def getpass(self, prompt="", stream=None) -> str:
    "Matches signature of [ipykernel.kernelbase.Kernel.getpass][]."
    return self.parent.input_request(str(prompt), password=True)

raw_input ¶

raw_input(prompt='') -> str

Matches signature of ipykernel.kernelbase.Kernel.raw_input.

Source code in src/async_kernel/kernel.py

def raw_input(self, prompt="") -> str:
    "Matches signature of [ipykernel.kernelbase.Kernel.raw_input][]."
    return self.parent.input_request(str(prompt), password=False)

kernel

KernelInterrupt ¶

Kernel ¶

help_links class-attribute instance-attribute ¶

supported_features class-attribute instance-attribute ¶

handle_in_shell_thread class-attribute instance-attribute ¶

handle_in_thread class-attribute instance-attribute ¶

callers class-attribute instance-attribute ¶

caller class-attribute instance-attribute ¶

debugger class-attribute instance-attribute ¶

comm_manager class-attribute instance-attribute ¶

active_execute_requests class-attribute instance-attribute ¶

kernel_info property ¶

shell property ¶

running async ¶

do_interrupt async ¶

apply_patches ¶

displayhook ¶

message_handler ¶

job ¶

send_reply ¶

get_shell ¶

subshell_id ¶

create_subshell ¶

protected ¶

kernel_info_request async ¶

comm_info_request async ¶

execute_request async ¶

complete_request async ¶

is_complete_request async ¶

inspect_request async ¶

history_request async ¶

comm_open async ¶

comm_msg async ¶

comm_close async ¶

interrupt_request async ¶

shutdown_request async ¶

debug_request async ¶

create_subshell_request async ¶

delete_subshell_request async ¶

list_subshell_request async ¶

get_parent ¶

do_complete async ¶

do_inspect async ¶

do_history async ¶

do_execute async ¶

getpass ¶

raw_input ¶

help_links `class-attribute` `instance-attribute` ¶

supported_features `class-attribute` `instance-attribute` ¶

handle_in_shell_thread `class-attribute` `instance-attribute` ¶

handle_in_thread `class-attribute` `instance-attribute` ¶

callers `class-attribute` `instance-attribute` ¶

caller `class-attribute` `instance-attribute` ¶

debugger `class-attribute` `instance-attribute` ¶

comm_manager `class-attribute` `instance-attribute` ¶

active_execute_requests `class-attribute` `instance-attribute` ¶

kernel_info `property` ¶

shell `property` ¶

running `async` ¶

do_interrupt `async` ¶

`job` ¶

`send_reply` ¶

`subshell_id` ¶

`protected` ¶

kernel_info_request `async` ¶

comm_info_request `async` ¶

execute_request `async` ¶

complete_request `async` ¶

is_complete_request `async` ¶

inspect_request `async` ¶

history_request `async` ¶

comm_open `async` ¶

comm_msg `async` ¶

comm_close `async` ¶

interrupt_request `async` ¶

shutdown_request `async` ¶

debug_request `async` ¶

create_subshell_request `async` ¶

delete_subshell_request `async` ¶

list_subshell_request `async` ¶

do_complete `async` ¶

do_inspect `async` ¶

do_history `async` ¶

do_execute `async` ¶