ipshell

Defines an IPython compatible shell.

Classes:

• NullContext –

  A context that does nothing that can be used where a context is expected.

• IPDisplayHook –

  Called by the kernel whenever the interpreter needs to display results.

• IPDisplayPublisher –

  A display publisher used by IPython.display.publish_display_data.

• IPHistoryManager –

  A class to organize history-related functionality in one place.

• IPShell –

  An IPython InteractiveShell implementation.

NullContext

A context that does nothing that can be used where a context is expected.

Used by:

• Reference ipshell  IPShell  builtin_trap

Source code in src/async_kernel/shell/ipshell.py

class NullContext:
    "A context that does nothing that can be used where a context is expected."

    __slots__ = ["weakref"]

    def __enter__(self) -> None:
        return

    def __exit__(self, type, value, traceback) -> Literal[False]:
        return False

IPDisplayHook

Bases: HasInterface, DisplayHook

Called by the kernel whenever the interpreter needs to display results.

The output is always published with msg_type="execute_result".

Returned by:

• Reference ipshell  IPShell  displayhook

Used by:

• Reference ipshell  IPShell  displayhook_class

Methods:

• __call__ –

  Publish the result.

Source code in src/async_kernel/shell/ipshell.py

class IPDisplayHook(HasInterface, DisplayHook):
    """
    Called by the kernel whenever the interpreter needs to display results.

    The output is always published with `msg_type="execute_result"`.
    """

    cache_size = traitlets.Int(1000, min=3).tag(config=True)
    do_full_cache = traitlets.Int(0).tag(config=True)

    _ = __ = ___ = ""

    def __init__(self, shell: IPShell) -> None:
        self._shell_ref = weakref.ref(shell)
        super(DisplayHook, self).__init__()

    @property
    def exec_result(self) -> ExecutionResult | None:  # pyright: ignore[reportImplicitOverride]
        return None  # pragma: no cover

    @exec_result.setter
    def exec_result(self, exec_result: ExecutionResult) -> None:
        pass  # pragma: no cover

    @property
    @override
    def shell(self) -> IPShell:
        return self._shell_ref()  # pyright: ignore[reportReturnType]

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

    @override
    def start_displayhook(self) -> None:
        pass  # pragma: no cover

    @override
    def write_output_prompt(self) -> None:
        pass  # pragma: no cover

    @override
    def write_format_data(self, format_dict, md_dict=None) -> None:
        pass  # pragma: no cover

    @override
    def finish_displayhook(self) -> None:
        pass  # pragma: no cover

    @override
    def quiet(self) -> bool:
        raise NotImplementedError  # pragma: no cover

    @override
    def __call__(self, result=None) -> None:
        """
        Publish the result.

        This is invoked every time the interpreter needs to print, and is
        activated by setting the variable sys.displayhook to it.
        """
        if result is not None and utils.show_result_enabled():
            format_dict, md_dict = self.shell.display_formatter.format(result)
            self.update_user_ns(result)
            if format_dict:
                content = {}
                content["execution_count"] = self.shell.execution_count
                content["data"] = format_dict
                content["metadata"] = md_dict
                self.log_output(format_dict)
                self.parent.iopub_send("execute_result", content=content)

__call__

__call__(result=None) -> None

Publish the result.

This is invoked every time the interpreter needs to print, and is activated by setting the variable sys.displayhook to it.

Source code in src/async_kernel/shell/ipshell.py

@override
def __call__(self, result=None) -> None:
    """
    Publish the result.

    This is invoked every time the interpreter needs to print, and is
    activated by setting the variable sys.displayhook to it.
    """
    if result is not None and utils.show_result_enabled():
        format_dict, md_dict = self.shell.display_formatter.format(result)
        self.update_user_ns(result)
        if format_dict:
            content = {}
            content["execution_count"] = self.shell.execution_count
            content["data"] = format_dict
            content["metadata"] = md_dict
            self.log_output(format_dict)
            self.parent.iopub_send("execute_result", content=content)

IPDisplayPublisher

Bases: HasInterface, DisplayPublisher

A display publisher used by IPython.display.publish_display_data.

Returned by:

• Reference ipshell  IPShell  display_pub

Used by:

• Reference ipshell  IPShell  display_pub_class

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/shell/ipshell.py

class IPDisplayPublisher(HasInterface, DisplayPublisher):
    """
    A display publisher used by [IPython.display.publish_display_data][].
    """

    _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.parent.msg(msg_type, content=content, parent=utils.get_parent_message())
        for hook in self._hooks:
            try:
                msg = hook(msg)
            except Exception:
                pass
            if msg is None:
                return
        self.parent.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.parent.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/shell/ipshell.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.parent.msg(msg_type, content=content, parent=utils.get_parent_message())
    for hook in self._hooks:
        try:
            msg = hook(msg)
        except Exception:
            pass
        if msg is None:
            return
    self.parent.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/shell/ipshell.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.parent.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/shell/ipshell.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)

IPHistoryManager

Bases: HasInterface[BaseInterface['IPShell']], HistoryManager

A class to organize history-related functionality in one place.

Returned by:

• Reference ipshell  IPShell  history_manager

Used by:

• Reference ipshell  IPShell  history_manager

Methods:

• __init__ –

  Create a new history manager associated with a shell instance.

Source code in src/async_kernel/shell/ipshell.py

class IPHistoryManager(HasInterface[BaseInterface["IPShell"]], HistoryManager):
    """
    A class to organize history-related functionality in one place.
    """

    @property
    @override
    def shell(self) -> IPShell:
        return self._shell_ref()  # pyright: ignore[reportReturnType]

    @override
    def __init__(self, *, shell: IPShell) -> None:
        """Create a new history manager associated with a shell instance."""
        self._shell_ref = weakref.ref(shell)
        hist_file = ":memory:" if not shell.is_mainshell or sys.platform == "emscripten" else ""

        super(HistoryManager, self).__init__(hist_file=hist_file)

        self.db_input_cache_lock = threading.Lock()
        self.db_output_cache_lock = threading.Lock()

        try:
            self.new_session()
        except OperationalError as e:
            self.log.exception(
                "Failed to create history session in %s. History will not be saved.", self.hist_file, exc_info=e
            )
            self.hist_file = ":memory:"

        self.using_thread = False
        if self.enabled and self.hist_file != ":memory:":
            self.save_thread = HistorySavingThread(self)
            utils.mark_thread_pydev_do_not_trace(self.save_thread)
            try:
                self.save_thread.start()
            except RuntimeError as e:
                self.log.exception(
                    "Failed to start history saving thread. History will not be saved.",
                    exc_info=e,
                )
                self.hist_file = ":memory:"
            else:
                self.using_thread = True
        else:
            self.save_thread = None
            if shell is not shell.kernel.main_shell:
                self.output_hist.update(shell.kernel.main_shell.history_manager.output_hist)

        self._instances.add(self)

    def stop(self) -> None:
        self.end_session()
        if thread := self.save_thread:
            thread.stop()
            thread.join()
            self.save_thread = None
        self._instances.discard(self)

__init__

__init__(*, shell: IPShell) -> None

Source code in src/async_kernel/shell/ipshell.py

@override
def __init__(self, *, shell: IPShell) -> None:
    """Create a new history manager associated with a shell instance."""
    self._shell_ref = weakref.ref(shell)
    hist_file = ":memory:" if not shell.is_mainshell or sys.platform == "emscripten" else ""

    super(HistoryManager, self).__init__(hist_file=hist_file)

    self.db_input_cache_lock = threading.Lock()
    self.db_output_cache_lock = threading.Lock()

    try:
        self.new_session()
    except OperationalError as e:
        self.log.exception(
            "Failed to create history session in %s. History will not be saved.", self.hist_file, exc_info=e
        )
        self.hist_file = ":memory:"

    self.using_thread = False
    if self.enabled and self.hist_file != ":memory:":
        self.save_thread = HistorySavingThread(self)
        utils.mark_thread_pydev_do_not_trace(self.save_thread)
        try:
            self.save_thread.start()
        except RuntimeError as e:
            self.log.exception(
                "Failed to start history saving thread. History will not be saved.",
                exc_info=e,
            )
            self.hist_file = ":memory:"
        else:
            self.using_thread = True
    else:
        self.save_thread = None
        if shell is not shell.kernel.main_shell:
            self.output_hist.update(shell.kernel.main_shell.history_manager.output_hist)

    self._instances.add(self)

IPShell

Bases: BaseShell, InteractiveShell

An IPython InteractiveShell implementation.

Used by:

• Reference ipshell  IPHistoryManager  __init__

Methods:

• apply_patches –

  Apply patches returning a callable to reverse the patches.

• init_hooks –

  Initialize hooks.

• run_line_magic_async –

  Call and awaits run_line_magic.

• run_cell_magic_async –

  Call and awaits run_cell_magic.

• system –

  Make a system call in a separate thread.

• run_code –

  Execute a code object.

• transform_cell_async –

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

• do_execute –

  Execute code in the shell's user_ns and global_ns.

• do_complete –
• is_complete –
• do_inspect –
• do_history –
• init_magics –

  Initialize magics.

• enable_matplotlib –

  Enable interactive matplotlib and inline figure support.

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.

• default_matplotlib_backends –
• compiler_class –
• prefilter_manager_class –
• displayhook_class –
• display_pub_class –
• display_formatter_class –
• configurables –

  Not used. Provided for compatibility.

• compile (Fixed[Self, XCachingCompiler]) –

  The compiler: provides a filename for a selection of code (cell).

• prefilter_manager (Fixed[Self, IPPrefilterManager]) –
• extension_manager (Fixed[Self, IPExtensionManager]) –

  A manager for loading extensions.

• builtin_trap –

  A nullcontext.

• displayhook (Fixed[Self, IPDisplayHook]) –

  An implementation of sys.displayhook.

• display_pub (Fixed[Self, IPDisplayPublisher]) –

  Used for publishing output generated by calls made to IPython.display.display which calls IPython.display.publish_display_data.

• display_formatter (Fixed[Self, IPDisplayFormatter]) –

  An object capable of transforming python objects to MIME content.

• history_manager (Fixed[Self, IPHistoryManager]) –

Source code in src/async_kernel/shell/ipshell.py

class IPShell(BaseShell, InteractiveShell):  # pyright: ignore[reportUnsafeMultipleInheritance, reportIncompatibleVariableOverride, reportIncompatibleMethodOverride]
    """
    An IPython InteractiveShell implementation.
    """

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

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

    default_matplotlib_backends = traitlets.List(["inline", "ipympl"]).tag(config=True)
    ""
    compiler_class = traitlets.Type(XCachingCompiler).tag(config=True)
    ""
    prefilter_manager_class = traitlets.Type(IPPrefilterManager).tag(config=True)
    ""
    displayhook_class = traitlets.Type(IPDisplayHook).tag(config=True)
    ""
    display_pub_class = traitlets.Type(IPDisplayPublisher).tag(config=True)
    ""
    display_formatter_class = traitlets.Type(IPDisplayFormatter).tag(config=True)
    ""

    configurables = Fixed(list)
    "Not used. Provided for compatibility."

    compile: Fixed[Self, XCachingCompiler] = Fixed(lambda c: c["owner"].compiler_class())
    "The compiler: provides a filename for a selection of code (cell)."

    prefilter_manager: Fixed[Self, IPPrefilterManager] = Fixed(
        lambda c: c["owner"].prefilter_manager_class(shell=c["owner"])
    )
    ""
    extension_manager: Fixed[Self, IPExtensionManager] = Fixed(lambda c: IPExtensionManager(shell=c["owner"]))
    "A manager for loading extensions."

    builtin_trap = Fixed(NullContext)
    """
    A nullcontext. 

    Builtins are not dynamically modified.
    """

    displayhook: Fixed[Self, IPDisplayHook] = Fixed(lambda c: c["owner"].displayhook_class(c["owner"]))  # pyright: ignore[reportIncompatibleMethodOverride]
    """
    An implementation of [sys.displayhook][]. 

    [async_kernel.kernel.Kernel.displayhook][] patches [sys.displayhook][] routing execution results to 
    the shell whose context the execution result occurred.
    """

    display_pub: Fixed[Self, IPDisplayPublisher] = Fixed(lambda c: c["owner"].display_pub_class())
    """
    Used for publishing output generated by calls made to [IPython.display.display][] which calls [IPython.display.publish_display_data][].
    """

    display_formatter: Fixed[Self, IPDisplayFormatter] = Fixed(lambda c: c["owner"].display_formatter_class())
    """
    An object capable of transforming python objects to MIME content.

    Notes:
        - Primarily used in [IPython.core.interactiveshell.InteractiveShell.user_expressions][].
        - [Ipython docs](https://ipython.readthedocs.io/en/stable/config/shell_mimerenderer.html).
    """

    history_manager: Fixed[Self, IPHistoryManager] = Fixed(lambda c: IPHistoryManager(shell=c["owner"]), mode="ignore")
    ""

    meta = Fixed(Struct)
    tempfiles = Fixed(list, mode="ignore")
    tempdirs = Fixed(list, mode="ignore")

    _main_mod_cache = Fixed(dict)
    _stop_on_error_pool: Fixed[Self, set[Pending[Any]]] = Fixed(set)

    # Disabled attributes
    loop_runner_map = None
    loop_runner = None
    autoindent = False
    call_pdb = Fixed(lambda _: None, mode="ignore")
    trio_runner = None

    # Disabled methods
    @override
    def init_prefilter(self) -> Never:
        raise MethodNotSupported  # pragma: no cover

    @override
    def init_create_namespaces(self, user_module=None, user_ns=None) -> Never:
        raise MethodNotSupported  # pragma: no cover

    @override
    def save_sys_module_state(self) -> Never:
        raise MethodNotSupported  # pragma: no cover

    @override
    def init_sys_modules(self) -> Never:
        raise MethodNotSupported  # pragma: no cover

    @override
    def init_history(self) -> Never:
        raise MethodNotSupported  # pragma: no cover

    @override
    def init_encoding(self) -> Never:
        raise MethodNotSupported  # pragma: no cover

    @override
    def init_user_ns(self) -> Never:
        raise MethodNotSupported  # pragma: no cover

    @override
    def init_instance_attrs(self) -> Never:
        raise MethodNotSupported  # pragma: no cover

    @override
    def init_payload(self) -> Never:
        raise MethodNotSupported  # pragma: no cover

    input_transformer_manager: traitlets.Instance[TransformerManager]

    @override
    def initialize(self):
        # Expects the kernel to be started.
        super().initialize()

        if leading_empty_lines in self.input_transformer_manager.cleanup_transforms:
            # leading_empty_lines can cause a line mismatch when debugging.
            self.input_transformer_manager.cleanup_transforms.remove(leading_empty_lines)

        self.init_ipython_dir(None)
        self.init_profile_dir(None)

        self.init_syntax_highlighting()
        self.init_hooks()
        self.init_events()
        self.init_pushd_popd_magic()
        self.init_logger()

        self.init_completer()

        self.init_traceback_handlers(((), None))
        self.init_prompts()

        self.init_magics()
        self.init_alias()
        self.init_logstart()

        self.user_ns_hidden.update(self.user_ns)
        self.events.trigger("shell_initialized", self)

    @override
    def apply_patches(self) -> Callable[[], None]:
        """Apply patches returning a callable to reverse the patches."""
        original = InteractiveShell.initialized, InteractiveShell.instance
        InteractiveShell.initialized = lambda: True
        InteractiveShell.instance = utils.get_ipython  # pyright: ignore[reportAttributeAccessIssue]

        restore_: Callable[[], None] = super().apply_patches()

        def restore():
            InteractiveShell.initialized, InteractiveShell.instance = original
            restore_()

        return restore

    @override
    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),
            "exit": self.exiter,
            "quit": self.exiter,
            "open": _modified_open_,
            "_": "",
            "__": "",
            "___": "",
        }

    @property
    @override
    def banner(self) -> str:
        return super(BaseShell, self).banner

    @traitlets.default("banner1")
    def _default_banner1(self) -> str:
        kernel_info = (
            f"async-kernel v{async_kernel.__version__} name:{self.parent.name!r} backend:{str(self.parent.backend)!r}"
        )
        return f"Python {sys.version}\n{kernel_info}\nIPython shell {IPython.core.release.version}\n"

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

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

    @override
    def init_builtins(self) -> None:
        super().init_builtins()
        if self.is_mainshell:
            builtins.__dict__["__IPYTHON__"] = True
            builtins.__dict__["display"] = display
            builtins.__dict__["get_ipython"] = utils.get_ipython
            builtins.__dict__["Caller"] = Caller

    @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.parent.iopub_send("display_data", content=data)
            else:
                self.parent.iopub_send("stream", content={"name": "stdout", "text": data})

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

    @contextmanager
    @override
    def _tee(self, channel: Literal["stdout", "stderr"]):
        yield

    @property
    @override
    def display_trap(self):
        show = True
        if msg := utils.get_parent_message():
            if "silent" in msg["content"]:
                show = not msg["content"]["silent"]
            if show and (code := msg["content"].get("code")) and str(code).strip().endswith(";"):
                show = False
        return utils.show_result(show)

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

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

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

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

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

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

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

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

        return Caller().to_thread(open_process)

    @override
    async def run_code(
        self, code_obj: CodeType, result: ExecutionResult | None = None, *, async_: bool = False
    ) -> bool:
        """Execute a code object.

        When an exception occurs, self.showtraceback() is called to display a traceback.

        Args:
            code_obj: A compiled code object, to be executed.
            result: An object to store exceptions that occur during execution.
            async_:  (Experimental) Attempt to run top-level asynchronous code in a default loop.

        Returns:
            False: successful execution.
            True: an error occurred.
        """
        try:
            if async_:
                await eval(code_obj, self.user_global_ns, self.user_ns)
            else:
                exec(code_obj, self.user_global_ns, self.user_ns)
        except self.custom_exceptions:
            etype, value, tb = sys.exc_info()
            if result is not None:
                result.error_in_exec = value
            self.CustomTB(etype, value, tb)
        except Exception as e:
            if result is not None:
                result.error_in_exec = e
            self.showtraceback(running_compiled_code=True)
        else:
            return False
        return True

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

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

    @override
    async def run_cell_async(
        self,
        raw_cell: str,
        store_history=False,
        silent=False,
        shell_futures=True,
        *,
        transformed_cell: str | None = None,
        preprocessing_exc_tuple: Any = None,
        cell_id=None,
    ) -> ExecutionResult:
        token = utils._cell_id_var.set(cell_id)  # pyright: ignore[reportPrivateUsage]
        result = None
        try:
            result = await super().run_cell_async(
                raw_cell=raw_cell,
                store_history=store_history,
                silent=silent,
                shell_futures=shell_futures,
                transformed_cell=transformed_cell,
                preprocessing_exc_tuple=preprocessing_exc_tuple,
                cell_id=cell_id,
            )
            return result  # noqa: RET504
        finally:
            self.events.trigger("post_execute")
            if not silent:
                self.events.trigger("post_run_cell", result)
            utils._cell_id_var.reset(token)  # pyright: ignore[reportPrivateUsage]

    @override
    async def do_execute(
        self,
        code: str = "",
        *,
        silent: bool = False,
        store_history: bool = False,
        user_expressions: dict[str, str] | None = None,
        allow_stdin: bool = False,
        stop_on_error: bool = False,
        cell_id: str | None = None,
        received_time: float = 0,
        tags: Iterable[str] = (),
        **_ignored,
    ) -> Content:
        """
        Execute code in the shell's user_ns and global_ns.
        """
        if received_time > 0 and (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)
            }
        if math.isnan(timeout := utils.get_tag_value(Tags.timeout, math.nan, tags=tags)):
            timeout = self.timeout

        if Tags.stop_on_error in tags:
            stop_on_error = utils.get_tag_value(Tags.stop_on_error, stop_on_error, tags=tags)
        elif Tags.raises_exception in tags or timeout:
            stop_on_error = False

        if silent:
            execution_count: int = self.execution_count
        else:
            execution_count = self._execution_count = self._execution_count + 1
            self.parent.iopub_send(
                msg_or_type="execute_input",
                content={"code": code, "execution_count": execution_count},
                ident=b"kernel.execute_input",
            )

        pen = Caller().call_soon(
            self.run_cell_async,
            raw_cell=code,
            store_history=store_history,
            silent=silent,
            transformed_cell=self.transform_cell_async(code),
            shell_futures=True,
            cell_id=cell_id,
        )
        err = result = None
        try:
            self.kernel.active_execute_requests.add(pen)
            pen.add_done_callback(self.kernel.active_execute_requests.discard)
            if stop_on_error:
                self._stop_on_error_pool.add(pen)
                pen.add_done_callback(self._stop_on_error_pool.discard)
            result = await pen.wait(timeout=timeout or None)
        except Exception as e:
            err = KernelInterrupt() if str(e) == self.kernel._interrupt_message else e  # pyright: ignore[reportPrivateUsage]
        else:
            err = result.error_before_exec or result.error_in_exec if result else KernelInterrupt()
            if not err and Tags.raises_exception in tags:
                msg = "An expected exception was not raised!"
                err = RuntimeError(msg)

        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:
                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 %s %s", self, pen)
                if stop_on_error:
                    for pen in self._stop_on_error_pool.copy():
                        pen.cancel("Stop on error cancellation")
        return content

    @override
    async def do_complete(self, code: str, cursor_pos: int | None = None) -> Content:
        ""

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

    @override
    async def is_complete(self, code: str) -> Content:
        ""
        status, indent_spaces = self.input_transformer_manager.check_complete(code)
        content = {"status": status}
        if isinstance(indent_spaces, int):
            content["indent"] = " " * indent_spaces
        return content

    @override
    async def do_inspect(self, code: str, cursor_pos: int = 0, detail_level: Literal[0, 1] = 0) -> Content:
        ""
        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

    @override
    async def do_history(
        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:
        ""
        history_manager = self.history_manager
        assert history_manager
        match hist_access_type:
            case "tail":
                hist = history_manager.get_tail(n=n, raw=raw, output=output, include_latest=False)
            case "range":
                hist = history_manager.get_range(session, start, stop, raw, output)
            case "search":
                hist = history_manager.search(pattern=pattern, raw=raw, output=output, n=n, unique=unique)
            case _:
                hist = []
        return {"history": list(hist), "status": "ok"}

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

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

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

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

        This takes the following steps:

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

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

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

        matplotlib_inline.backend_inline.configure_inline_support(self, backend)

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

        return gui, backend

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

    @override
    def stop(self, *, force=False) -> None:
        if self.protected and not force:
            return
        super().stop(force=force)
        self.configurables.clear()
        self.user_ns.clear()
        self.history_manager.stop()
        try:
            self.atexit_operations()
        except AttributeError:
            pass

timeout class-attribute instance-attribute

timeout = traitlets.CFloat(0.0).tag(config=True)

A timeout in seconds to complete execute requests.

stop_on_error_time_offset class-attribute instance-attribute

stop_on_error_time_offset = traitlets.Float(0.0).tag(config=True)

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

default_matplotlib_backends class-attribute instance-attribute

default_matplotlib_backends = traitlets.List(['inline', 'ipympl']).tag(config=True)

compiler_class class-attribute instance-attribute

compiler_class = traitlets.Type(XCachingCompiler).tag(config=True)

prefilter_manager_class class-attribute instance-attribute

prefilter_manager_class = traitlets.Type(IPPrefilterManager).tag(config=True)

displayhook_class class-attribute instance-attribute

displayhook_class = traitlets.Type(IPDisplayHook).tag(config=True)

display_pub_class class-attribute instance-attribute

display_pub_class = traitlets.Type(IPDisplayPublisher).tag(config=True)

display_formatter_class class-attribute instance-attribute

display_formatter_class = traitlets.Type(IPDisplayFormatter).tag(config=True)

configurables class-attribute instance-attribute

configurables = Fixed(list)

Not used. Provided for compatibility.

compile class-attribute instance-attribute

compile: Fixed[Self, XCachingCompiler] = Fixed(lambda c: c['owner'].compiler_class())

The compiler: provides a filename for a selection of code (cell).

prefilter_manager class-attribute instance-attribute

prefilter_manager: Fixed[Self, IPPrefilterManager] = Fixed(
    lambda c: c["owner"].prefilter_manager_class(shell=c["owner"])
)

extension_manager class-attribute instance-attribute

extension_manager: Fixed[Self, IPExtensionManager] = Fixed(
    lambda c: IPExtensionManager(shell=c["owner"])
)

A manager for loading extensions.

builtin_trap class-attribute instance-attribute

builtin_trap = Fixed(NullContext)

A nullcontext.

Builtins are not dynamically modified.

displayhook class-attribute instance-attribute

displayhook: Fixed[Self, IPDisplayHook] = Fixed(
    lambda c: c["owner"].displayhook_class(c["owner"])
)

An implementation of sys.displayhook.

async_kernel.kernel.Kernel.displayhook patches sys.displayhook routing execution results to the shell whose context the execution result occurred.

display_pub class-attribute instance-attribute

display_pub: Fixed[Self, IPDisplayPublisher] = Fixed(
    lambda c: c["owner"].display_pub_class()
)

Used for publishing output generated by calls made to IPython.display.display which calls IPython.display.publish_display_data.

display_formatter class-attribute instance-attribute

display_formatter: Fixed[Self, IPDisplayFormatter] = Fixed(
    lambda c: c["owner"].display_formatter_class()
)

An object capable of transforming python objects to MIME content.

Notes

• Primarily used in IPython.core.interactiveshell.InteractiveShell.user_expressions.
• Ipython docs.

history_manager class-attribute instance-attribute

history_manager: Fixed[Self, IPHistoryManager] = Fixed(
    lambda c: IPHistoryManager(shell=c["owner"]), mode="ignore"
)

apply_patches

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

Apply patches returning a callable to reverse the patches.

Source code in src/async_kernel/shell/ipshell.py

@override
def apply_patches(self) -> Callable[[], None]:
    """Apply patches returning a callable to reverse the patches."""
    original = InteractiveShell.initialized, InteractiveShell.instance
    InteractiveShell.initialized = lambda: True
    InteractiveShell.instance = utils.get_ipython  # pyright: ignore[reportAttributeAccessIssue]

    restore_: Callable[[], None] = super().apply_patches()

    def restore():
        InteractiveShell.initialized, InteractiveShell.instance = original
        restore_()

    return restore

init_hooks

init_hooks() -> None

Initialize hooks.

Source code in src/async_kernel/shell/ipshell.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.parent.iopub_send("display_data", content=data)
        else:
            self.parent.iopub_send("stream", content={"name": "stdout", "text": data})

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

run_line_magic_async async

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

Call and awaits run_line_magic.

Source code in src/async_kernel/shell/ipshell.py

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

run_cell_magic_async async

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

Call and awaits run_cell_magic.

Source code in src/async_kernel/shell/ipshell.py

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

system

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

Make a system call in a separate thread.

Parameters:

• cmd

  (list[str] | str) –

  Passed as the first argument 'command' when calling anyio.open_process.

• stderr_to_stdout

  (bool, default: False ) –

  Send stderr output to stdout.

• **kwargs

  (Any, default: {} ) –

  Keyword arguments are passed to anyio.open_process.

Tip

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

Source code in src/async_kernel/shell/ipshell.py

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

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

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

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

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

    return Caller().to_thread(open_process)

run_code async

run_code(
    code_obj: CodeType, result: ExecutionResult | None = None, *, async_: bool = False
) -> bool

Execute a code object.

When an exception occurs, self.showtraceback() is called to display a traceback.

Parameters:

• code_obj

  (CodeType) –

  A compiled code object, to be executed.

• result

  (ExecutionResult | None, default: None ) –

  An object to store exceptions that occur during execution.

• async_

  (bool, default: False ) –

  (Experimental) Attempt to run top-level asynchronous code in a default loop.

Returns:

• False ( bool ) –

  successful execution.

• True ( bool ) –

  an error occurred.

Source code in src/async_kernel/shell/ipshell.py

@override
async def run_code(
    self, code_obj: CodeType, result: ExecutionResult | None = None, *, async_: bool = False
) -> bool:
    """Execute a code object.

    When an exception occurs, self.showtraceback() is called to display a traceback.

    Args:
        code_obj: A compiled code object, to be executed.
        result: An object to store exceptions that occur during execution.
        async_:  (Experimental) Attempt to run top-level asynchronous code in a default loop.

    Returns:
        False: successful execution.
        True: an error occurred.
    """
    try:
        if async_:
            await eval(code_obj, self.user_global_ns, self.user_ns)
        else:
            exec(code_obj, self.user_global_ns, self.user_ns)
    except self.custom_exceptions:
        etype, value, tb = sys.exc_info()
        if result is not None:
            result.error_in_exec = value
        self.CustomTB(etype, value, tb)
    except Exception as e:
        if result is not None:
            result.error_in_exec = e
        self.showtraceback(running_compiled_code=True)
    else:
        return False
    return True

transform_cell_async

transform_cell_async(raw_cell: str) -> str

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

Source code in src/async_kernel/shell/ipshell.py

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

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

do_execute async

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

Execute code in the shell's user_ns and global_ns.

Source code in src/async_kernel/shell/ipshell.py

@override
async def do_execute(
    self,
    code: str = "",
    *,
    silent: bool = False,
    store_history: bool = False,
    user_expressions: dict[str, str] | None = None,
    allow_stdin: bool = False,
    stop_on_error: bool = False,
    cell_id: str | None = None,
    received_time: float = 0,
    tags: Iterable[str] = (),
    **_ignored,
) -> Content:
    """
    Execute code in the shell's user_ns and global_ns.
    """
    if received_time > 0 and (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)
        }
    if math.isnan(timeout := utils.get_tag_value(Tags.timeout, math.nan, tags=tags)):
        timeout = self.timeout

    if Tags.stop_on_error in tags:
        stop_on_error = utils.get_tag_value(Tags.stop_on_error, stop_on_error, tags=tags)
    elif Tags.raises_exception in tags or timeout:
        stop_on_error = False

    if silent:
        execution_count: int = self.execution_count
    else:
        execution_count = self._execution_count = self._execution_count + 1
        self.parent.iopub_send(
            msg_or_type="execute_input",
            content={"code": code, "execution_count": execution_count},
            ident=b"kernel.execute_input",
        )

    pen = Caller().call_soon(
        self.run_cell_async,
        raw_cell=code,
        store_history=store_history,
        silent=silent,
        transformed_cell=self.transform_cell_async(code),
        shell_futures=True,
        cell_id=cell_id,
    )
    err = result = None
    try:
        self.kernel.active_execute_requests.add(pen)
        pen.add_done_callback(self.kernel.active_execute_requests.discard)
        if stop_on_error:
            self._stop_on_error_pool.add(pen)
            pen.add_done_callback(self._stop_on_error_pool.discard)
        result = await pen.wait(timeout=timeout or None)
    except Exception as e:
        err = KernelInterrupt() if str(e) == self.kernel._interrupt_message else e  # pyright: ignore[reportPrivateUsage]
    else:
        err = result.error_before_exec or result.error_in_exec if result else KernelInterrupt()
        if not err and Tags.raises_exception in tags:
            msg = "An expected exception was not raised!"
            err = RuntimeError(msg)

    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:
            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 %s %s", self, pen)
            if stop_on_error:
                for pen in self._stop_on_error_pool.copy():
                    pen.cancel("Stop on error cancellation")
    return content

do_complete async

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

Source code in src/async_kernel/shell/ipshell.py

@override
async def do_complete(self, code: str, cursor_pos: int | None = None) -> Content:
    ""

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

is_complete async

is_complete(code: str) -> Content

Source code in src/async_kernel/shell/ipshell.py

@override
async def is_complete(self, code: str) -> Content:
    ""
    status, indent_spaces = self.input_transformer_manager.check_complete(code)
    content = {"status": status}
    if isinstance(indent_spaces, int):
        content["indent"] = " " * indent_spaces
    return content

do_inspect async

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

Source code in src/async_kernel/shell/ipshell.py

@override
async def do_inspect(self, code: str, cursor_pos: int = 0, detail_level: Literal[0, 1] = 0) -> Content:
    ""
    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

do_history async

do_history(
    *,
    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

Source code in src/async_kernel/shell/ipshell.py

@override
async def do_history(
    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:
    ""
    history_manager = self.history_manager
    assert history_manager
    match hist_access_type:
        case "tail":
            hist = history_manager.get_tail(n=n, raw=raw, output=output, include_latest=False)
        case "range":
            hist = history_manager.get_range(session, start, stop, raw, output)
        case "search":
            hist = history_manager.search(pattern=pattern, raw=raw, output=output, n=n, unique=unique)
        case _:
            hist = []
    return {"history": list(hist), "status": "ok"}

init_magics

init_magics() -> None

Initialize magics.

Source code in src/async_kernel/shell/ipshell.py

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

enable_matplotlib

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

Enable interactive matplotlib and inline figure support.

This takes the following steps:

1. select the appropriate matplotlib backend
2. set up matplotlib for interactive use with that backend
3. 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/shell/ipshell.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