Asyncshell

async_kernel.asyncshell

Classes:

• AsyncDisplayHook –

  A displayhook subclass that publishes data using ZeroMQ.

• AsyncDisplayPublisher –

  A display publisher that publishes data using a ZeroMQ PUB socket.

• AsyncInteractiveShell –

  An IPython InteractiveShell modified to work with Async kernel.

AsyncDisplayHook

A displayhook subclass that publishes data using ZeroMQ.

This is intended to work with an InteractiveShell instance. It sends a dict of different representations of the object.

Methods:

• finish_displayhook –

  Finish up all displayhook activities.

• start_displayhook –

  Start the display hook.

• write_format_data –

  Write format data to the message.

• write_output_prompt –

  Write the output prompt.

finish_displayhook

finish_displayhook() -> None

Finish up all displayhook activities.

Source code in src/async_kernel/asyncshell.py

@override
def finish_displayhook(self) -> None:
    """Finish up all displayhook activities."""
    if self.content:
        self.kernel.iopub_send("display_data", content=self.content)
        self.content = {}

start_displayhook

start_displayhook() -> None

Start the display hook.

Source code in src/async_kernel/asyncshell.py

@override
def start_displayhook(self) -> None:
    """Start the display hook."""
    self.content = {}

write_format_data

write_format_data(format_dict, md_dict=None) -> None

Write format data to the message.

Source code in src/async_kernel/asyncshell.py

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

write_output_prompt

write_output_prompt() -> None

Write the output prompt.

Source code in src/async_kernel/asyncshell.py

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

AsyncDisplayPublisher

A display publisher that publishes data using a ZeroMQ PUB socket.

Methods:

• clear_output –

  Clear output associated with the current execution (cell).

• publish –

  Publish a display-data message.

clear_output

clear_output(wait: bool = False) -> None

Clear output associated with the current execution (cell).

Parameters:

• wait

  (bool, default: False ) –

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

Source code in src/async_kernel/asyncshell.py

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

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

publish

publish(
    data: Content,
    metadata: dict | None = None,
    *,
    transient: dict | None = None,
    update: bool = False,
    **kwargs,
) -> None

Publish a display-data message.

Parameters:

• data

  (Content) –

  A mime-bundle dict, keyed by mime-type.

• metadata

  (dict | None, default: None ) –

  Metadata associated with the data.

• transient

  (dict | None, default: None ) –

  Transient data that may only be relevant during a live display, such as display_id. Transient data should not be persisted to documents.

• update

  (bool, default: False ) –

  If True, send an update_display_data message instead of display_data.

Reference

Source code in src/async_kernel/asyncshell.py

@override
def publish(  # pyright: ignore[reportIncompatibleMethodOverride]
    self,
    data: Content,
    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)
    """
    utils.get_kernel().iopub_send(
        msg_or_type="update_display_data" if update else "display_data",
        content={"data": data, "metadata": metadata or {}, "transient": transient or {}} | kwargs,
        ident=self.topic,
    )

AsyncInteractiveShell

An IPython InteractiveShell modified to work with Async kernel.

Notable differences

• All execute requests are run asynchronously.

• Supports a soft timeout with the metadata {"timeout":}¹.

• Not all features are support (see "not-supported" features listed below).

---

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

Methods:

• init_magics –

  Initialize magics.

• run_cell_async –

  Run a complete IPython cell asynchronously.

Attributes:

• autoindent –

  not-supported

• debug –

  not-supported

• execute_request_timeout –

  A timeout in seconds to complete execute requests.

• kernel (Kernel) –

  The current kernel.

• loop_runner –

  not-supported

• loop_runner_map –

  not-supported

• readline_use –

  not-supported

• run_cell –

  not-supported

autoindent class-attribute instance-attribute

autoindent = False

not-supported

debug class-attribute instance-attribute

debug = None

not-supported

execute_request_timeout class-attribute instance-attribute

execute_request_timeout = CFloat(default_value=None, allow_none=True)

A timeout in seconds to complete execute requests.

kernel property

kernel: Kernel

The current kernel.

loop_runner class-attribute instance-attribute

loop_runner = None

not-supported

loop_runner_map class-attribute instance-attribute

loop_runner_map = None

not-supported

readline_use class-attribute instance-attribute

readline_use = False

not-supported

run_cell class-attribute instance-attribute

run_cell = None

not-supported

init_magics

init_magics() -> None

Initialize magics.

Source code in src/async_kernel/asyncshell.py

@override
def init_magics(self) -> None:
    """Initialize magics."""
    super().init_magics()
    self.register_magics(KernelMagics)

run_cell_async async

run_cell_async(
    raw_cell: str,
    store_history=False,
    silent=False,
    shell_futures=True,
    *,
    transformed_cell: str | None = None,
    preprocessing_exc_tuple: tuple | None = None,
    cell_id: str | None = None,
) -> ExecutionResult

Run a complete IPython cell asynchronously.

This function runs execute requests for the kernel wrapping InteractiveShell.

Source code in src/async_kernel/asyncshell.py

@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: tuple | None = None,
    cell_id: str | None = None,
) -> ExecutionResult:
    """
    Run a complete IPython cell asynchronously.

    This function runs [execute requests][async_kernel.Kernel.execute_request] for the kernel
    wrapping [InteractiveShell][IPython.core.interactiveshell.InteractiveShell.run_cell_async].
    """
    with anyio.fail_after(delay=utils.get_execute_request_timeout()):
        result: ExecutionResult = 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,
        )
    self.events.trigger("post_execute")
    if not silent:
        self.events.trigger("post_run_cell", result)
    return result

KernelMagics

Extra magics for async kernel.

Methods:

• callers –

  Print a table of Callers, indicating its status including: -running - protected - on the current thread.

• connect_info –

  Print information for connecting other clients to this kernel.

callers

callers(_) -> None

Print a table of Callers, indicating its status including: -running - protected - on the current thread.

Source code in src/async_kernel/asyncshell.py

@line_magic
def callers(self, _) -> None:
    "Print a table of [Callers][async_kernel.Caller], indicating its status including:  -running - protected - on the current thread."
    lines = ["\t".join(["Running", "Protected", "\t", "Name"]), "─" * 70]
    for caller in Caller.all_callers(running_only=False):
        symbol = "   ✓" if caller.running else "   ✗"
        current_thread: Literal["← current thread", ""] = "← current thread" if caller is Caller() else ""
        protected = "   🔐" if caller.protected else ""
        lines.append("\t".join([symbol, protected, "", caller.thread.name, current_thread]))
    print(*lines, sep="\n")

connect_info

connect_info(_) -> None

Print information for connecting other clients to this kernel.

Source code in src/async_kernel/asyncshell.py

@line_magic
def connect_info(self, _) -> None:
    """Print information for connecting other clients to this kernel."""
    kernel = utils.get_kernel()
    connection_file = pathlib.Path(kernel.connection_file)
    # if it's in the default dir, truncate to basename
    if jupyter_runtime_dir() == str(connection_file.parent):
        connection_file = connection_file.name
    info = kernel.get_connection_info()
    print(
        json.dumps(info, indent=2, default=json_default),
        "Paste the above JSON into a file, and connect with:\n"
        + "    $> jupyter <app> --existing <file>\n"
        + "or, if you are local, you can connect with just:\n"
        + f"    $> jupyter <app> --existing {connection_file}\n"
        + "or even just:\n"
        + "    $> jupyter <app> --existing\n"
        + "if this is the most recent Jupyter kernel you have started.",
    )