euporie.core.kernel.jupyter

Contain the main class for a notebook file.

Functions

cast(typ, val)	Cast a value to a type.
uuid4()	Generate a random UUID.

Classes

BaseKernel(kernel_tab[, default_callbacks, ...])	Abstract base class for euporie kernels.
JupyterKernel(kernel_tab[, ...])	Run a notebook kernel and communicates with it asynchronously.
KernelInfo(name, display_name, type, kind, ...)	Named tuple representing a launchable kernel.
MsgCallbacks	Typed dictionary for named message callbacks.
UPath(*args[, protocol])
defaultdict	defaultdict(default_factory=None, /, [...]) --> dict with default factory
partial(func, /, *args, **keywords)	Create a new function with partial application of the given arguments and keywords.

class euporie.core.kernel.jupyter.JupyterKernel(kernel_tab: KernelTab, default_callbacks: MsgCallbacks | None = None, allow_stdin: bool = False, *, kernel_name: str | None = None, connection_file: Path | None = None, **kwargs: Any)

Bases: BaseKernel

Run a notebook kernel and communicates with it asynchronously.

Has the ability to run itself in it’s own thread.

comm_info(target_name: str | None = None) → None

Request information about the current comms.

complete(source: str, cursor_pos: int) → list[dict]

Request code completions from the kernel.

Parameters:

• source – The code string to retrieve completions for

• cursor_pos – The position of the cursor in the code string

Returns:

A list of dictionaries defining completion entries. The dictionaries contain text (the completion text), start_position (the stating position of the completion text), and optionally display_meta (a string containing additional data about the completion type)

async complete_async(source: str, cursor_pos: int, timeout: int = 60) → list[dict]

Request code completions from the kernel, asynchronously.

history(pattern: str = '', n: int = 1, hist_access_type: str = 'search') → list[tuple[int, int, str]] | None

Retrieve history from the kernel.

Parameters:

• pattern – The pattern to search for

• n – the number of history items to return

• hist_access_type – How to access the history (‘range’, ‘tail’ or ‘search’)

Returns:

A list of history items, consisting of tuples (session, line_number, input)

async history_async(pattern: str = '', n: int = 1, hist_access_type: str = 'search', timeout: int = 60) → list[tuple[int, int, str]] | None

Retrieve history from the kernel asynchronously.

property id: str | None

Get the ID of the current kernel.

info(set_kernel_info: Callable[[dict[str, Any]], None] | None = None, set_status: Callable[[str], None] | None = None) → None

Request information about the kernel.

input(text: str) → None

Send input to the kernel.

inspect(source: str, cursor_pos: int, detail_level: int = 0, timeout: int = 2, callback: Callable[[dict[str, Any]], None] | None = None) → dict[str, Any]

Request code inspection from the kernel.

Parameters:

• source – The code string to retrieve completions for

• cursor_pos – The position of the cursor in the code string

• detail_level – Level of detail for the inspection (0-2)

• timeout – Number of seconds to wait for inspection results

• callback – A function to run when the inspection result arrives. The result is passed as an argument.

Returns:

A string containing useful information about the code at the current cursor position

async inspect_async(source: str, cursor_pos: int, detail_level: int = 0, timeout: int = 2) → dict[str, Any]

Retrieve introspection string from the kernel asynchronously.

interrupt() → None

Interrupt the kernel.

This is run in the main thread rather than on the event loop in the kernel’s thread, because otherwise we would have to wait for currently running tasks on the kernel’s event loop to finish.

is_complete(source: str, timeout: int | float = 0.1, callback: Callable[[dict[str, Any]], None] | None = None) → dict[str, Any]

Request code completeness status from the kernel.

Parameters:

• source – The code string to check the completeness status of

• timeout – How long to wait for a kernel response

• wait – Whether to wait for the response

• callback – A function to run when the inspection result arrives. The result is passed as an argument.

Returns:

A string describing the completeness status

async is_complete_async(source: str, timeout: int | float = 0.1) → dict[str, Any]

Ask the kernel to determine if code is complete asynchronously.

kc_comm(comm_id: str, data: dict[str, Any]) → str

Send a comm message on the shell channel.

property missing: bool

Return True if the requested kernel is not found.

async monitor_status() → None

Regularly monitor the kernel status.

on_iopub_clear_output(rsp: dict[str, Any], own: bool) → None

Call callbacks for an iopub clear output response.

on_iopub_comm_close(rsp: dict[str, Any], own: bool) → None

Call callbacks for an iopub comm close response.

on_iopub_comm_msg(rsp: dict[str, Any], own: bool) → None

Call callbacks for an iopub comm message response.

on_iopub_comm_open(rsp: dict[str, Any], own: bool) → None

Call callbacks for an comm open response.

on_iopub_display_data(rsp: dict[str, Any], own: bool) → None

Call callbacks for an iopub display data response.

on_iopub_error(rsp: dict[str, Any], own: bool) → None

Call callbacks for an iopub error response.

on_iopub_execute_input(rsp: dict[str, Any], own: bool) → None

Call callbacks for an iopub execute input response.

on_iopub_execute_result(rsp: dict[str, Any], own: bool) → None

Call callbacks for an iopub execute result response.

on_iopub_shutdown_reply(rsp: dict[str, Any], own: bool) → None

Handle iopub shutdown reply messages.

on_iopub_status(rsp: dict[str, Any], own: bool) → None

Call callbacks for an iopub status response.

on_iopub_stream(rsp: dict[str, Any], own: bool) → None

Call callbacks for an iopub stream response.

on_iopub_update_display_data(rsp: dict[str, Any], own: bool) → None

Call callbacks for an iopub update display data response.

on_shell_complete_reply(rsp: dict[str, Any], own: bool) → None

Call callbacks for a shell completion reply response.

on_shell_execute_reply(rsp: dict[str, Any], own: bool) → None

Call callbacks for a shell execute reply response.

on_shell_history_reply(rsp: dict[str, Any], own: bool) → None

Call callbacks for a shell history reply response.

on_shell_inspect_reply(rsp: dict[str, Any], own: bool) → None

Call callbacks for a shell inspection reply response.

on_shell_is_complete_reply(rsp: dict[str, Any], own: bool) → None

Call callbacks for a shell completeness reply response.

on_shell_kernel_info_reply(rsp: dict[str, Any], own: bool) → None

Call callbacks for a shell kernel info response.

on_shell_status(rsp: dict[str, Any], own: bool) → None

Call set_execution_count callback for a shell status response.

on_stdin_input_request(rsp: dict[str, Any], own: bool) → None

Call get_input callback for a stdin input request message.

on_unhandled(channel: str, rsp: dict[str, Any], own: bool) → None

Report unhandled messages to the debug log.

async poll(channel: str) → None

Poll for messages on a channel, and signal when they arrive.

Parameters:

channel – The name of the channel to get messages from

async post_start() → None

Wait for the kernel to become ready.

restart(wait: bool = False, cb: Callable | None = None) → None

Restart the current kernel.

async restart_async() → None

Restart the kernel asyncchronously.

run(source: str, wait: bool = False, callback: Callable[..., None] | None = None, **callbacks: Callable[..., Any]) → None

Run a cell using the notebook kernel and process the responses.

async run_async(source: str, **local_callbacks: Unpack[MsgCallbacks]) → None

Run the code cell and and set the response callbacks, optionally waiting.

shutdown(wait: bool = False, cb: Callable | None = None) → None

Shutdown the kernel.

This is intended to be run when the notebook is closed: the BaseKernel cannot be restarted after this.

Parameters:

• wait – Whether to block until shutdown completes

• cb – Callback run after shutdown completes

async shutdown_async() → None

Shut down the kernel and close the event loop if running in a thread.

property spec: dict[str, str]

The kernelspec metadata for the current kernel instance.

start(cb: Callable | None = None, wait: bool = False, timeout: int = 10) → None

Start the kernel.

Parameters:

• cb – An optional callback to run after the kernel has started

• wait – If True, block until the kernel has started

• timeout – How long to wait until failure is assumed

async start_async() → None

Start the kernel asynchronously and set its status.

property status: str

Retrieve the current kernel status.

Returns:

The kernel status

stop(cb: Callable | None = None, wait: bool = False) → None

Stop the current kernel.

Parameters:

• cb – An optional callback to run when the kernel has stopped.

• wait – If True, wait for the kernel to become idle, otherwise the kernel is interrupted before it is stopped

async stop_async(cb: Callable[[], Any] | None = None) → None

Stop the kernel asynchronously.

classmethod variants() → list[KernelInfo]

Return available kernel specifications.

wait_for_status(status: str = 'idle') → None

Block until the kernel reaches a given status value.