euporie.core.kernel.local

Local Python interpreter kernel implementation.

Functions

`cast`(typ, val)	Cast a value to a type.
`get_app`()	Get the current active (running) Application.
`get_display_data`(obj)	Get display data and metadata for an object.
`highlight`(code, lexer, formatter[, outfile])	This is the most high-level highlighting function.
`to_thread`(func, /, args, *kwargs)	Asynchronously run function func in a separate thread.
`update_wrapper`(wrapper, wrapped[, assigned, ...])	Update a wrapper function to look like the wrapped function

Classes

`BaseHook`(kernel)	Base class providing access to thread-specific callbacks.
`BaseKernel`(kernel_tab[, default_callbacks, ...])	Abstract base class for euporie kernels.
`DisplayGlobal`(kernel)	A display() function that dispatches to thread-specific callbacks.
`DisplayHook`(kernel)	Hook for sys.displayhook that dispatches to thread-specific callbacks.
`HookManager`(kernel)	Context manager for hooking stdout/stderr/displayhook.
`InputBuiltin`(kernel[, is_password])	Hook for input and getpass that dispatches to thread-specific callbacks.
`KernelInfo`(name, display_name, type, kind, ...)	Named tuple representing a launchable kernel.
`LocalPythonKernel`(kernel_tab[, ...])	Run code in a local Python interpreter.
`MsgCallbacks`	Typed dictionary for named message callbacks.
`Path`(args, *kwargs)	PurePath subclass that can make system calls.
`Python3TracebackLexer`	alias of `PythonTracebackLexer`
`StreamWrapper`(name, kernel)	Hook for stdout/stderr that dispatches to thread-specific callbacks.
`Terminal256Formatter`(**options)	Format tokens with ANSI color sequences, for output in a 256-color terminal or console.

class euporie.core.kernel.local.BaseHook(kernel: LocalPythonKernel)

Bases: object

Base class providing access to thread-specific callbacks.

property callbacks: MsgCallbacks: Get callbacks for current thread.

class euporie.core.kernel.local.DisplayGlobal(kernel: LocalPythonKernel)

Bases: BaseHook

A display() function that dispatches to thread-specific callbacks.

This class implements the global display() function used to show rich output in notebooks. It routes display calls to the appropriate output callbacks.

property callbacks: MsgCallbacks: Get callbacks for current thread.

class euporie.core.kernel.local.DisplayHook(kernel: LocalPythonKernel)

Bases: BaseHook

Hook for sys.displayhook that dispatches to thread-specific callbacks.

property callbacks: MsgCallbacks: Get callbacks for current thread.

class euporie.core.kernel.local.HookManager(kernel: LocalPythonKernel)

Bases: object

Context manager for hooking stdout/stderr/displayhook.

class euporie.core.kernel.local.InputBuiltin(kernel: LocalPythonKernel, is_password: bool = False)

Bases: BaseHook

Hook for input and getpass that dispatches to thread-specific callbacks.

property callbacks: MsgCallbacks: Get callbacks for current thread.

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

Bases: BaseKernel

Run code in a local Python interpreter.

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.

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.

Parameters:

set_kernel_info – Callback to set kernel info
set_status – Callback to set kernel status

init_venv() → None

Add the current venv to sys.path so we can import modules from it.

This isn’t perfect: it doesn’t use the Python interpreter with which the virtualenv was built, and it ignores the –no-site-packages option. A warning will appear suggesting the user installs euporie in the virtualenv or to use a Jupyter kernel, but for many cases, it probably works well enough.

Adapted from IPython.

input(text: str) → None

Send input to the kernel.

Parameters:: text – The input text to provide

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.

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.

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(wait: bool = False, cb: Callable | None = None) → None: Restart the kernel.

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 local interpreter.

showtraceback(filename: str, cb: Callable[[dict[str, Any], bool], None] | None) → None: Format and display tracebacks for exceptions.

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(wait: bool = False) → None: Shutdown the kernel.

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 local interpreter.

property status: str

Retrieve the current kernel status.

Returns:: The kernel status

stop(cb: Callable | None = None, wait: bool = False) → None: Stop the local interpreter.

async stop_async(cb: Callable | 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.

class euporie.core.kernel.local.StreamWrapper(name: str, kernel: LocalPythonKernel)

Bases: BaseHook

Hook for stdout/stderr that dispatches to thread-specific callbacks.

property buffer: str: Get the thread-local buffer for this stream.

property callbacks: MsgCallbacks: Get callbacks for current thread.

flush() → None: Flush any buffered content.

write(text: str) → int: Write output using callback for current thread.

euporie.core.kernel.local.get_display_data(obj: Any) → tuple[dict[str, Any], dict[str, Any]]

Get display data and metadata for an object.

Parameters:: obj – Object to get display data for
Returns:: Tuple of (data, metadata) dictionaries