euporie.core.kernel.base

Base class for euporie kernels.

Functions

NamedTuple(typename[, fields])

Typed version of namedtuple.

TypedDict(typename[, fields, total])

A simple typed namespace.

abstractmethod(funcobj)

A decorator indicating abstract methods.

get_loop()

Create or return the conversion IO loop.

overload(func)

Decorator for overloaded functions/methods.

Classes

ABC()

Helper class that provides a standard way to create an ABC using inheritance.

BaseKernel(kernel_tab[, default_callbacks, ...])

Abstract base class for euporie kernels.

KernelInfo(name, display_name, type, kind, ...)

Named tuple representing a launchable kernel.

MsgCallbacks

Typed dictionary for named message callbacks.

NoKernel(kernel_tab[, default_callbacks, ...])

A None kernel.
class euporie.core.kernel.base.BaseKernel(kernel_tab: KernelTab, default_callbacks: MsgCallbacks | None = None, allow_stdin: bool = False, **kwargs: Any)

Bases: ABC

Abstract base class for euporie kernels.

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

Request information about the current comms.

Does nothing by default.

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)

abstractmethod async complete_async(source: str, cursor_pos: int) → list[dict]

Get code completions 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 = 1) → list[tuple[int, int, str]] | None

Retrieve history from the kernel asynchronously.

abstract property id: str | None

Return the kernel ID.

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

Request information about the kernel.

abstractmethod 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

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

Get code inspection/documentation asynchronously.

abstractmethod interrupt() → None

Interrupt the kernel.

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

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

Check if code is complete asynchronously.

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

By default kernels do not implement COMM communication.

abstract property missing: bool

Return whether the kernel is missing.

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

Restart the current kernel.

abstractmethod async restart_async() → None

Restart the kernel asynchronously.

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

Execute code in the kernel.

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

Execute code in the kernel asynchronously.

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

abstractmethod async shutdown_async() → None

Shutdown the kernel asynchronously.

abstract 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

abstractmethod async start_async() → None

Start the kernel.

property status: str

Retrieve the current kernel status.

Returns:

The kernel status

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

Stop the kernel.

async stop_async(cb: Callable | None = None) → None

Stop the kernel asynchronously.

classmethod variants() → list[KernelInfo]

Return a list of parameterized variants of this kernel.

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

Block until the kernel reaches a given status value.

class euporie.core.kernel.base.KernelInfo(name: str, display_name: str, type: type[BaseKernel], kind: Literal['new', 'existing'], factory: KernelFactory)

Bases: NamedTuple

Named tuple representing a launchable kernel.

count(value, /)

Return number of occurrences of value.

display_name: str

Alias for field number 1

factory: KernelFactory

Alias for field number 4

index(value, start=0, stop=9223372036854775807, /)

Return first index of value.

Raises ValueError if the value is not present.

kind: Literal['new', 'existing']

Alias for field number 3

name: str

Alias for field number 0

type: type[BaseKernel]

Alias for field number 2

class euporie.core.kernel.base.MsgCallbacks

Bases: TypedDict

Typed dictionary for named message callbacks.

add_input: Callable[[dict[str, Any], bool], None] | None
add_output: Callable[[dict[str, Any], bool], None] | None
ask_exit: Callable[[bool], None] | None
clear()

Remove all items from the dict.

clear_output: Callable[[bool], None] | None
completeness_status: Callable[[dict[str, Any]], None] | None
copy()

Return a shallow copy of the dict.

dead: Callable[[], None] | None
done: Callable[[dict[str, Any]], None] | None
edit_magic: Callable[[str, int], None] | None
classmethod fromkeys(iterable, value=None, /)

Create a new dictionary with keys from iterable and values set to value.

get(key, default=None, /)

Return the value for key if key is in the dictionary, else default.

get_input: Callable[[str, bool], None] | None
items()

Return a set-like object providing a view on the dict’s items.

keys()

Return a set-like object providing a view on the dict’s keys.

page: Callable[[list[dict], int], None] | None
pop(k[, d]) → v, remove specified key and return the corresponding value.

If the key is not found, return the default if given; otherwise, raise a KeyError.

popitem()

Remove and return a (key, value) pair as a 2-tuple.

Pairs are returned in LIFO (last-in, first-out) order. Raises KeyError if the dict is empty.

set_execution_count: Callable[[int], None] | None
set_kernel_info: Callable[[dict[str, Any]], None] | None
set_metadata: Callable[[tuple[str, ...], Any], None] | None
set_next_input: Callable[[str, bool], None] | None
set_status: Callable[[str], None] | None
setdefault(key, default=None, /)

Insert key with a value of default if key is not in the dictionary.

Return the value for key if key is in the dictionary, else default.

update([E, ]**F) → None.  Update D from dict/iterable E and F.

If E is present and has a .keys() method, then does: for k in E: D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]

values()

Return an object providing a view on the dict’s values.

class euporie.core.kernel.base.NoKernel(kernel_tab: KernelTab, default_callbacks: MsgCallbacks | None = None, allow_stdin: bool = False, **kwargs: Any)

Bases: BaseKernel

A None kernel.

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

Request information about the current comms.

Does nothing by default.

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) → list[dict]

Get code completions 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 = 1) → list[tuple[int, int, str]] | None

Retrieve history from the kernel asynchronously.

property id: str | None

Return the kernel ID.

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]

Get code inspection/documentation asynchronously.

interrupt() → None

Interrupt the kernel.

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]

Check if code is complete asynchronously.

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

By default kernels do not implement COMM communication.

property missing: bool

Return whether the kernel is missing.

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

Restart the current kernel.

async restart_async() → None

Restart the kernel asynchronously.

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

Execute code in the kernel.

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

Execute code in the kernel asynchronously.

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

Shutdown the kernel asynchronously.

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.

property status: str

Retrieve the current kernel status.

Returns:

The kernel status

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

Stop the kernel.

async stop_async(cb: Callable | None = None) → None

Stop the kernel asynchronously.

classmethod variants() → list[KernelInfo]

Return a list of parameterized variants of this kernel.

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

Block until the kernel reaches a given status value.

euporie.core.kernel.base.get_loop() → AbstractEventLoop

Create or return the conversion IO loop.

The loop will be running on a separate thread.