Skip to content

async_kernel.asyncshell

Classes:

async_kernel.asyncshell.AsyncDisplayHook

Bases: DisplayHook

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:

start_displayhook 

start_displayhook() -> None

Start the display hook.

Source code in src/async_kernel/asyncshell.py 
49
50
51
52
@override
def start_displayhook(self) -> None:
    """Start the display hook."""
    self.content = {}

write_output_prompt 

write_output_prompt() -> None

Write the output prompt.

Source code in src/async_kernel/asyncshell.py 
54
55
56
57
@override
def write_output_prompt(self) -> None:
    """Write the output prompt."""
    self.content["execution_count"] = self.prompt_count

write_format_data 

write_format_data(format_dict, md_dict=None) -> None

Write format data to the message.

Source code in src/async_kernel/asyncshell.py 
59
60
61
62
63
@override
def write_format_data(self, format_dict, md_dict=None) -> None:
    """Write format data to the message."""
    self.content["data"] = format_dict
    self.content["metadata"] = md_dict

finish_displayhook 

finish_displayhook() -> None

Finish up all displayhook activities.

Source code in src/async_kernel/asyncshell.py 
65
66
67
68
69
70
@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 = {}

async_kernel.asyncshell.AsyncDisplayPublisher 

AsyncDisplayPublisher(shell=None, *args, **kwargs)

Bases: DisplayPublisher

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

Methods:

  • publish –

    Publish a display-data message.

  • clear_output –

    Clear output associated with the current execution (cell).

  • register_hook –

    Register a hook for when publish is called.

Source code in src/async_kernel/asyncshell.py 
78
79
80
def __init__(self, shell=None, *args, **kwargs) -> None:
    super().__init__(shell, *args, **kwargs)
    self._hooks = []

publish 

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

Publish a display-data message.

Parameters:

  • data

    (dict[str, Any]) –

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

  • metadata

    (dict | None, default: None ) –

    Metadata associated with the data.

  • transient

    (dict | None, default: None ) –

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

  • update

    (bool, default: False ) –

    If True, send an update_display_data message instead of display_data.

Reference

Source code in src/async_kernel/asyncshell.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@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 = utils.get_kernel().session.msg(msg_type, content, parent=utils.get_parent())  # pyright: ignore[reportArgumentType]
    for hook in self._hooks:
        try:
            msg = hook(msg)
        except Exception:
            pass
        if msg is None:
            return
    utils.get_kernel().iopub_send(msg)

clear_output 

clear_output(wait: bool = False) -> None

Clear output associated with the current execution (cell).

Parameters:

  • wait

    (bool, default: False ) –

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

Source code in src/async_kernel/asyncshell.py 
116
117
118
119
120
121
122
123
124
125
126
@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)

register_hook 

register_hook(hook: Callable[[dict], dict | None]) -> None

Register a hook for when publish is called.

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

Source code in src/async_kernel/asyncshell.py 
128
129
130
131
132
133
134
def register_hook(self, hook: Callable[[dict], dict | None]) -> 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)

async_kernel.asyncshell.AsyncInteractiveShell

Bases: InteractiveShell

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":}1.

  • 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:

Attributes:

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.

run_cell class-attribute instance-attribute 

run_cell = None

not-supported

loop_runner_map class-attribute instance-attribute 

loop_runner_map = None

not-supported

loop_runner class-attribute instance-attribute 

loop_runner = None

not-supported

debug class-attribute instance-attribute 

debug = None

not-supported

readline_use class-attribute instance-attribute 

readline_use = False

not-supported

autoindent class-attribute instance-attribute 

autoindent = False

not-supported

kernel property 

kernel: Kernel

The current kernel.

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 
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
@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

init_magics 

init_magics() -> None

Initialize magics.

Source code in src/async_kernel/asyncshell.py 
295
296
297
298
299
@override
def init_magics(self) -> None:
    """Initialize magics."""
    super().init_magics()
    self.register_magics(KernelMagics)

async_kernel.asyncshell.KernelMagics

Bases: Magics

Extra magics for async kernel.

Methods:

  • connect_info –

    Print information for connecting other clients to this kernel.

  • callers –

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

connect_info 

connect_info(_) -> None

Print information for connecting other clients to this kernel.

Source code in src/async_kernel/asyncshell.py 
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
@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.",
    )

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 
330
331
332
333
334
335
336
337
338
339
@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")