euporie.core.kernel

Contain the main class for a notebook file.

Functions

`TypedDict`(typename[, fields, total])	A simple typed namespace.
`jupyter_path`(*subdirs)	Return a list of directories to search for data files
`jupyter_runtime_dir`()	Return the runtime dir for transient jupyter files.
`uuid4`()	Generate a random UUID.

Classes

`AsyncKernelManager`(**kwargs)	An async kernel manager.
`EuporieKernelManager`(**kwargs)	Kernel Manager subclass.
`KPF`	alias of `jupyter_client.provisioning.factory.KernelProvisionerFactory`
`Kernel`(kernel_tab[, threaded, allow_stdin, ...])	Run a notebook kernel and communicates with it asynchronously.
`KernelManager`(**kwargs)	Manages a single kernel in a subprocess on this host.
`LocalProvisioner`(**kwargs)	`LocalProvisioner` is a concrete class of ABC `KernelProvisionerBase` and is the out-of-box default implementation used when no kernel provisioner is specified in the kernel specification (`kernel.json`).
`LoggingLocalProvisioner`(**kwargs)	A Jupyter kernel provisionser which logs kernel output.
`MsgCallbacks`	Typed dictionary for named message callbacks.
`UPath`(args, *kwargs)
`defaultdict`	defaultdict(default_factory=None, /, [...]) --> dict with default factory

Exceptions

NoSuchKernel(name)

An error raised when there is no kernel of a give name.

class euporie.core.kernel.EuporieKernelManager(**kwargs: Any)

Bases: jupyter_client.manager.AsyncKernelManager

Kernel Manager subclass.

jupyter_client replaces a plain python command with the current executable, but this is not desirable if the client is running in its own prefix (e.g. with pipx). We work around this here.

See https://github.com/jupyter/jupyter_client/issues/949

add_restart_callback(callback: Callable, event: str = 'restart') → None: Register a callback to be called when a kernel is restarted

add_traits(**traits): Dynamically add trait attributes to the HasTraits instance.

autorestart: Bool: Should we autorestart the kernel if it dies.

blocking_class: A trait whose value must be a subclass of a specified class.

blocking_client(): Make a blocking client connected to my kernel

cache_ports: Bool: True if the MultiKernelManager should cache ports for this KernelManager instance

classmethod class_config_rst_doc() → str

Generate rST documentation for this class’ config options.

Excludes traits defined on parent classes.

classmethod class_config_section(classes: Optional[Sequence[type[traitlets.traitlets.HasTraits]]] = None) → str

Get the config section for this class.

Parameters: classes (list, optional) – The list of other classes in the config file. Used to reduce redundant information.

classmethod class_get_help(inst: traitlets.traitlets.HasTraits | None = None) → str

Get the help string for this class in ReST format.

If inst is given, its current trait values will be used in place of class defaults.

classmethod class_get_trait_help(trait: traitlets.traitlets.TraitType[Any, Any], inst: traitlets.traitlets.HasTraits | None = None, helptext: str | None = None) → str

Get the helptext string for a single trait.

Parameters

inst – If given, its current trait values will be used in place of the class default.
helptext – If not given, uses the help attribute of the current trait.

classmethod class_own_trait_events(name)

Get a dict of all event handlers defined on this class, not a parent.

Works like event_handlers, except for excluding traits from parents.

classmethod class_own_traits(**metadata)

Get a dict of all the traitlets defined on this class, not a parent.

Works like class_traits, except for excluding traits from parents.

classmethod class_print_help(inst: traitlets.traitlets.HasTraits | None = None) → None: Get the help string for a single trait and print it.

classmethod class_trait_names(**metadata)

Get a list of all the names of this class’ traits.

This method is just like the trait_names() method, but is unbound.

classmethod class_traits(**metadata)

Get a dict of all the traits of this class. The dictionary is keyed on the name and the values are the TraitType objects.

This method is just like the traits() method, but is unbound.

The TraitTypes returned don’t know anything about the values that the various HasTrait’s instances are holding.

The metadata kwargs allow functions to be passed in which filter traits based on metadata values. The functions should take a single value as an argument and return a boolean. If any function returns False, then the trait is not included in the output. If a metadata key doesn’t exist, None will be passed to the function.

cleanup_connection_file() → None

Cleanup connection file if we wrote it

Will not raise if the connection file was already removed somehow.

cleanup_ipc_files() → None: Cleanup ipc files if we wrote them.

cleanup_random_ports() → None

Forgets randomly assigned port numbers and cleans up the connection file.

Does nothing if no port numbers have been randomly assigned. In particular, does nothing unless the transport is tcp.

async cleanup_resources(restart: bool = False) → None: Clean up resources when the kernel is shut down

client(**kwargs: Any) → jupyter_client.asynchronous.client.AsyncKernelClient: Get a client for the manager.

client_class: DottedObjectName: A string holding a valid dotted object name in Python, such as A.b3._c

client_factory: Type: A trait whose value must be a subclass of a specified class.

config

A trait whose value must be an instance of a specified class.

The value can also be an instance of a subclass of the specified class.

Subclasses can declare default classes by overriding the klass attribute

connect_control(identity: Optional[bytes] = None) → zmq.sugar.socket.Socket: return zmq Socket connected to the Control channel

connect_hb(identity: Optional[bytes] = None) → zmq.sugar.socket.Socket: return zmq Socket connected to the Heartbeat channel

connect_iopub(identity: Optional[bytes] = None) → zmq.sugar.socket.Socket: return zmq Socket connected to the IOPub channel

connect_shell(identity: Optional[bytes] = None) → zmq.sugar.socket.Socket: return zmq Socket connected to the Shell channel

connect_stdin(identity: Optional[bytes] = None) → zmq.sugar.socket.Socket: return zmq Socket connected to the StdIn channel

connection_file

kernel-<pid>.json]

This file will contain the IP, ports, and authentication key needed to connect clients to this kernel. By default, this file will be created in the security dir of the current profile, but can be specified by absolute path.

Type: JSON file in which to store connection info [default

context: Instance

A trait whose value must be an instance of a specified class.

The value can also be an instance of a subclass of the specified class.

Subclasses can declare default classes by overriding the klass attribute

control_port

random]

Type: set the control (ROUTER) port [default

property cross_validation_lock

A contextmanager for running a block with our cross validation lock set to True.

At the end of the block, the lock’s value is restored to its value prior to entering the block.

data_dir: Union[str, Unicode]: A trait for unicode strings.

async finish_shutdown(waittime: Optional[float] = None, pollinterval: float = 0.1, restart: bool = False) → None

Wait for kernel shutdown, then kill process if it doesn’t shutdown.

This does not send shutdown requests - use request_shutdown() first.

format_kernel_cmd(extra_arguments: list[str] | None = None) → list[str]: Replace templated args (e.g. {connection_file}).

get_connection_info(session: bool = False) → Dict[str, Union[int, str, bytes]]

Return the connection info as a dict

Parameters: session (bool [default: False]) – If True, return our session object will be included in the connection info. If False (default), the configuration parameters of our session object will be included, rather than the session object itself.
Returns: connect_info – dictionary of connection information.
Return type: dict

property has_kernel: bool: Has a kernel process been started that we are actively managing.

has_trait(name): Returns True if the object has a trait with the specified name.

hb_port

random]

Type: set the heartbeat port [default

hold_trait_notifications()

Context manager for bundling trait change notifications and cross validation.

Use this when doing multiple trait assignments (init, config), to avoid race conditions in trait notifiers requesting other trait values. All trait notifications will fire after all values have been assigned.

async interrupt_kernel() → None

Interrupts the kernel by sending it a signal.

Unlike signal_kernel, this operation is well supported on all platforms.

iopub_port

random]

Type: set the iopub (PUB) port [default

ip: Set the kernel’s IP address [default localhost]. If the IP address is something other than localhost, then Consoles on other machines will be able to connect to the Kernel, so be careful!

property ipykernel: bool

async is_alive() → bool: Is the kernel process still running?

kernel_id: t.Union[str, Unicode]: A trait for unicode strings.

kernel_name: t.Union[str, Unicode]: A trait for unicode strings.

property kernel_spec: Optional[jupyter_client.kernelspec.KernelSpec]

kernel_spec_manager: Instance

A trait whose value must be an instance of a specified class.

The value can also be an instance of a subclass of the specified class.

Subclasses can declare default classes by overriding the klass attribute

load_connection_file(connection_file: Optional[str] = None) → None

Load connection info from JSON dict in self.connection_file.

Parameters: connection_file (unicode, optional) – Path to connection file to load. If unspecified, use self.connection_file

load_connection_info(info: Dict[str, Union[int, str, bytes]]) → None

Load connection info from a dict containing connection info.

Typically this data comes from a connection file and is called by load_connection_file.

Parameters: info (dict) – Dictionary containing connection_info. See the connection_file spec for details.

log: Logger or LoggerAdapter instance

notify_change(change): Notify observers of a change event

observe(handler: Callable[[...], Any], names: Union[traitlets.utils.sentinel.Sentinel, str, Iterable[traitlets.utils.sentinel.Sentinel | str]] = traitlets.All, type: traitlets.utils.sentinel.Sentinel | str = 'change') → None

Setup a handler to be called when a trait changes.

This is used to setup dynamic notifications of trait changes.

Parameters

handler (callable) – A callable that is called when a trait changes. Its signature should be handler(change), where change is a dictionary. The change dictionary at least holds a ‘type’ key. * type: the type of notification. Other keys may be passed depending on the value of ‘type’. In the case where type is ‘change’, we also have the following keys: * owner : the HasTraits instance * old : the old value of the modified trait attribute * new : the new value of the modified trait attribute * name : the name of the modified trait attribute.
names (list, str, All) – If names is All, the handler will apply to all traits. If a list of str, handler will apply to all names in the list. If a str, the handler will apply just to that name.
type (str, All (default: 'change')) – The type of notification to filter by. If equal to All, then all notifications are passed to the observe handler.

on_trait_change(handler=None, name=None, remove=False)

DEPRECATED: Setup a handler to be called when a trait changes.

This is used to setup dynamic notifications of trait changes.

Static handlers can be created by creating methods on a HasTraits subclass with the naming convention ‘_[traitname]_changed’. Thus, to create static handler for the trait ‘a’, create the method _a_changed(self, name, old, new) (fewer arguments can be used, see below).

If remove is True and handler is not specified, all change handlers for the specified name are uninstalled.

Parameters

handler (callable, None) – A callable that is called when a trait changes. Its signature can be handler(), handler(name), handler(name, new), handler(name, old, new), or handler(name, old, new, self).
name (list, str, None) – If None, the handler will apply to all traits. If a list of str, handler will apply to all names in the list. If a str, the handler will apply just to that name.
remove (bool) – If False (the default), then install the handler. If True then unintall it.

property owns_kernel: bool

parent

A trait whose value must be an instance of a specified class.

The value can also be an instance of a subclass of the specified class.

Subclasses can declare default classes by overriding the klass attribute

property ports: List[int]

async post_start_kernel(**kw: Any) → None

Performs any post startup tasks relative to the kernel.

Parameters: **kw (optional) – keyword arguments that were used in the kernel process’s launch.

async pre_start_kernel(**kw: Any) → Tuple[List[str], Dict[str, Any]]

Prepares a kernel for startup in a separate process.

If random ports (port=0) are being used, this method must be called before the channels are created.

Parameters: **kw (optional) – keyword arguments that are passed down to build the kernel_cmd and launching the kernel (e.g. Popen kwargs).

provisioner: t.Optional[KernelProvisionerBase] = None

property ready: Union[concurrent.futures._base.Future, _asyncio.Future]: A future that resolves when the kernel process has started for the first time

remove_restart_callback(callback: Callable, event: str = 'restart') → None: Unregister a callback to be called when a kernel is restarted

async request_shutdown(restart: bool = False) → None: Send a shutdown request via control channel

async restart_kernel(now: bool = False, newports: bool = False, **kw: Any) → None

Restarts a kernel with the arguments that were used to launch it.

Parameters

now (bool, optional) –
If True, the kernel is forcefully restarted immediately, without having a chance to do any cleanup action. Otherwise the kernel is given 1s to clean up before a forceful restart is issued.

In all cases the kernel is restarted, the only difference is whether it is given a chance to perform a clean shutdown or not.
newports (bool, optional) – If the old kernel was launched with random ports, this flag decides whether the same ports and connection file will be used again. If False, the same ports and connection file are used. This is the default. If True, new random port numbers are chosen and a new connection file is written. It is still possible that the newly chosen random port numbers happen to be the same as the old ones.
**kw (optional) – Any options specified here will overwrite those used to launch the kernel.

classmethod section_names() → list[str]: return section names as a list

session

A trait whose value must be an instance of a specified class.

The value can also be an instance of a subclass of the specified class.

Subclasses can declare default classes by overriding the klass attribute

set_trait(name, value): Forcibly sets trait attribute, including read-only attributes.

setup_instance(**kwargs): This is called before self.__init__ is called.

shell_port

random]

Type: set the shell (ROUTER) port [default

async shutdown_kernel(now: bool = False, restart: bool = False) → None

Attempts to stop the kernel process cleanly.

This attempts to shutdown the kernels cleanly by:

Sending it a shutdown message over the control channel.
If that fails, the kernel is shutdown forcibly by sending it a signal.

Parameters

now (bool) – Should the kernel be forcible killed now. This skips the first, nice shutdown attempt.
restart (bool) – Will this kernel be restarted after it is shutdown. When this is True, connection files will not be cleaned up.

shutdown_wait_time: Float: Time to wait for a kernel to terminate before killing it, in seconds. When a shutdown request is initiated, the kernel will be immediately sent an interrupt (SIGINT), followedby a shutdown_request message, after 1/2 of shutdown_wait_time`it will be sent a terminate (SIGTERM) request, and finally at the end of `shutdown_wait_time will be killed (SIGKILL). terminate and kill may be equivalent on windows. Note that this value can beoverridden by the in-use kernel provisioner since shutdown times mayvary by provisioned environment.

shutting_down: bool = False

async signal_kernel(signum: int) → None

Sends a signal to the process group of the kernel (this usually includes the kernel and any subprocesses spawned by the kernel).

Note that since only SIGTERM is supported on Windows, this function is only useful on Unix systems.

async start_kernel(**kw: Any) → None

Starts a kernel on this host in a separate process.

If random ports (port=0) are being used, this method must be called before the channels are created.

Parameters: **kw (optional) – keyword arguments that are passed down to build the kernel_cmd and launching the kernel (e.g. Popen kwargs).

start_restarter() → None: Start the kernel restarter.

stdin_port

random]

Type: set the stdin (ROUTER) port [default

stop_restarter() → None: Stop the kernel restarter.

trait_defaults(*names, **metadata)

Return a trait’s default value or a dictionary of them

Notes

Dynamically generated default values may depend on the current state of the object.

classmethod trait_events(name=None)

Get a dict of all the event handlers of this class.

Parameters: name (str (default: None)) – The name of a trait of this class. If name is None then all the event handlers of this class will be returned instead.
Return type: The event handlers associated with a trait name, or all event handlers.

trait_has_value(name)

Returns True if the specified trait has a value.

This will return false even if getattr would return a dynamically generated default value. These default values will be recognized as existing only after they have been generated.

Example

class MyClass(HasTraits):
    i = Int()

mc = MyClass()
assert not mc.trait_has_value("i")
mc.i # generates a default value
assert mc.trait_has_value("i")

trait_metadata(traitname, key, default=None): Get metadata values for trait by key.

trait_names(**metadata): Get a list of all the names of this class’ traits.

trait_values(**metadata)

A dict of trait names and their values.

The metadata kwargs allow functions to be passed in which filter traits based on metadata values. The functions should take a single value as an argument and return a boolean. If any function returns False, then the trait is not included in the output. If a metadata key doesn’t exist, None will be passed to the function.

Return type: A dict of trait names and their values.

Notes

Trait values are retrieved via getattr, any exceptions raised by traits or the operations they may trigger will result in the absence of a trait value in the result dict.

traits(**metadata)

Get a dict of all the traits of this class. The dictionary is keyed on the name and the values are the TraitType objects.

The TraitTypes returned don’t know anything about the values that the various HasTrait’s instances are holding.

The metadata kwargs allow functions to be passed in which filter traits based on metadata values. The functions should take a single value as an argument and return a boolean. If any function returns False, then the trait is not included in the output. If a metadata key doesn’t exist, None will be passed to the function.

transport: An enum of strings where the case should be ignored.

unobserve(handler: Callable[[...], Any], names: Union[traitlets.utils.sentinel.Sentinel, str, Iterable[traitlets.utils.sentinel.Sentinel | str]] = traitlets.All, type: traitlets.utils.sentinel.Sentinel | str = 'change') → None

Remove a trait change handler.

This is used to unregister handlers to trait change notifications.

Parameters

handler (callable) – The callable called when a trait attribute changes.
names (list, str, All (default: All)) – The names of the traits for which the specified handler should be uninstalled. If names is All, the specified handler is uninstalled from the list of notifiers corresponding to all changes.
type (str or All (default: 'change')) – The type of notification to filter by. If All, the specified handler is uninstalled from the list of notifiers corresponding to all types.

unobserve_all(name: str | Any = traitlets.All) → None: Remove trait change handlers of any type for the specified name. If name is not specified, removes all trait notifiers.

update_config(config: traitlets.config.loader.Config) → None: Update config and load the new values

write_connection_file(**kwargs: Any) → None: Write connection info to JSON dict in self.connection_file.

class euporie.core.kernel.Kernel(kernel_tab: KernelTab, threaded: bool = True, allow_stdin: bool = False, default_callbacks: MsgCallbacks | None = None, connection_file: Path | None = None)

Bases: object

Run a notebook kernel and communicates with it asynchronously.

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

change(name: str | None, connection_file: Path | None = None, cb: Callable | None = None) → None

Change the kernel.

Parameters

name – The name of the kernel to change to
connection_file – The path to the connection file to use
cb – Callback to run once restarted

comm_info(target_name: str | None = None) → None: Request information about the current comms.

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

Request code completions from the kernel.

Parameters

code – 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_(code: 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_(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: 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.

inspect(code: str, cursor_pos: int, callback: Callable[[dict[str, Any]], None] | None = None) → str

Request code inspection from the kernel.

Parameters

code – The code string to retrieve completions for
cursor_pos – The position of the cursor in the code string
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_(code: 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(code: str, timeout: int | float = 0.1, wait: bool = False, callback: Callable[[dict[str, Any]], None] | None = None) → dict[str, Any]

Request code completeness status from the kernel.

Parameters

code – 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_(code: 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.

on_iopub_clear_output(rsp: dict[str, Any]) → None: Call callbacks for an iopub clear output response.

on_iopub_comm_close(rsp: dict[str, Any]) → None: Call callbacks for an iopub comm close response.

on_iopub_comm_msg(rsp: dict[str, Any]) → None: Call callbacks for an iopub comm message response.

on_iopub_comm_open(rsp: dict[str, Any]) → None: Call callbacks for an comm open response.

on_iopub_display_data(rsp: dict[str, Any]) → None: Call callbacks for an iopub display data response.

on_iopub_error(rsp: dict[str, dict[str, Any]]) → None: Call callbacks for an iopub error response.

on_iopub_execute_input(rsp: dict[str, Any]) → None: Call callbacks for an iopub execute input response.

on_iopub_execute_result(rsp: dict[str, Any]) → None: Call callbacks for an iopub execute result response.

on_iopub_status(rsp: dict[str, Any]) → None: Call callbacks for an iopub status response.

on_iopub_stream(rsp: dict[str, Any]) → None: Call callbacks for an iopub stream response.

on_iopub_update_display_data(rsp: dict[str, Any]) → None: Call callbacks for an iopub update display data response.

on_shell_complete_reply(rsp: dict[str, Any]) → None: Call callbacks for a shell completion reply response.

on_shell_execute_reply(rsp: dict[str, Any]) → None: Call callbacks for a shell execute reply response.

on_shell_history_reply(rsp: dict[str, Any]) → None: Call callbacks for a shell history reply response.

on_shell_inspect_reply(rsp: dict[str, Any]) → None: Call callbacks for a shell inspection reply response.

on_shell_is_complete_reply(rsp: dict[str, Any]) → None: Call callbacks for a shell completeness reply response.

on_shell_kernel_info_reply(rsp: dict[str, Any]) → None: Call callbacks for a shell kernel info response.

on_shell_status(rsp: dict[str, Any]) → None: Call set_execution_count callback for a shell status response.

on_stdin_input_request(rsp: dict[str, Any]) → None: Call get_input callback for a stdin input request message.

on_unhandled(channel: str, rsp: dict[str, Any]) → 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_() → 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_(source: str, get_input: Callable[[str, bool], None] | None = None, set_execution_count: Callable[[int], None] | None = None, add_output: Callable[[dict[str, Any]], None] | None = None, clear_output: Callable[[bool], None] | None = None, done: Callable[[dict[str, Any]], None] | None = None, set_metadata: Callable[[tuple[str, ...], Any], None] | None = None, set_status: Callable[[str], None] | None = None) → None: Run the code cell and and set the response callbacks, optionally waiting.

shutdown(wait: bool = False) → None

Shutdown the kernel and close the kernel’s thread.

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

Parameters: wait – Whether to block until shutdown completes

async shutdown_() → None: Shut down the kernel and close the event loop if running in a thread.

property specs: dict[str, dict]: Return a list of available kernelspecs.

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_() → None: Start the kernel asynchronously and set its status.

property status: str

Retrieve the current kernel status.

Trigger a kernel life status check when retrieved

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_(cb: Callable[[], Any] | None = None) → None: Stop the kernel asynchronously.

wait_for_status(status: str = 'idle') → None: Block until the kernel reaches a given status value.

class euporie.core.kernel.LoggingLocalProvisioner(**kwargs: Any)

Bases: jupyter_client.provisioning.local_provisioner.LocalProvisioner

A Jupyter kernel provisionser which logs kernel output.

add_traits(**traits): Dynamically add trait attributes to the HasTraits instance.

classmethod class_config_rst_doc() → str

Generate rST documentation for this class’ config options.

Excludes traits defined on parent classes.

classmethod class_config_section(classes: Optional[Sequence[type[traitlets.traitlets.HasTraits]]] = None) → str

Get the config section for this class.

Parameters: classes (list, optional) – The list of other classes in the config file. Used to reduce redundant information.

classmethod class_get_help(inst: traitlets.traitlets.HasTraits | None = None) → str

Get the help string for this class in ReST format.

If inst is given, its current trait values will be used in place of class defaults.

classmethod class_get_trait_help(trait: traitlets.traitlets.TraitType[Any, Any], inst: traitlets.traitlets.HasTraits | None = None, helptext: str | None = None) → str

Get the helptext string for a single trait.

Parameters

inst – If given, its current trait values will be used in place of the class default.
helptext – If not given, uses the help attribute of the current trait.

classmethod class_own_trait_events(name)

Get a dict of all event handlers defined on this class, not a parent.

Works like event_handlers, except for excluding traits from parents.

classmethod class_own_traits(**metadata)

Get a dict of all the traitlets defined on this class, not a parent.

Works like class_traits, except for excluding traits from parents.

classmethod class_print_help(inst: traitlets.traitlets.HasTraits | None = None) → None: Get the help string for a single trait and print it.

classmethod class_trait_names(**metadata)

Get a list of all the names of this class’ traits.

This method is just like the trait_names() method, but is unbound.

classmethod class_traits(**metadata)

Get a dict of all the traits of this class. The dictionary is keyed on the name and the values are the TraitType objects.

This method is just like the traits() method, but is unbound.

The TraitTypes returned don’t know anything about the values that the various HasTrait’s instances are holding.

The metadata kwargs allow functions to be passed in which filter traits based on metadata values. The functions should take a single value as an argument and return a boolean. If any function returns False, then the trait is not included in the output. If a metadata key doesn’t exist, None will be passed to the function.

async cleanup(restart: bool = False) → None: Clean up the resources used by the provisioner and optionally restart.

config

A trait whose value must be an instance of a specified class.

The value can also be an instance of a subclass of the specified class.

Subclasses can declare default classes by overriding the klass attribute

connection_info: KernelConnectionInfo = {}

property cross_validation_lock

A contextmanager for running a block with our cross validation lock set to True.

At the end of the block, the lock’s value is restored to its value prior to entering the block.

async get_provisioner_info() → Dict: Captures the base information necessary for persistence relative to this instance.

get_shutdown_wait_time(recommended: float = 5.0) → float

Returns the time allowed for a complete shutdown. This may vary by provisioner.

This method is called from KernelManager.finish_shutdown() during the graceful phase of its kernel shutdown sequence.

The recommended value will typically be what is configured in the kernel manager.

get_stable_start_time(recommended: float = 10.0) → float

Returns the expected upper bound for a kernel (re-)start to complete. This may vary by provisioner.

The recommended value will typically be what is configured in the kernel restarter.

property has_process: bool

Returns true if this provisioner is currently managing a process.

This property is asserted to be True immediately following a call to the provisioner’s launch_kernel() method.

has_trait(name): Returns True if the object has a trait with the specified name.

hold_trait_notifications()

Context manager for bundling trait change notifications and cross validation.

Use this when doing multiple trait assignments (init, config), to avoid race conditions in trait notifiers requesting other trait values. All trait notifications will fire after all values have been assigned.

ip = None

kernel_id: Union[str, Unicode]: A trait for unicode strings.

kernel_spec: Any

A trait whose value must be an instance of a specified class.

The value can also be an instance of a subclass of the specified class.

Subclasses can declare default classes by overriding the klass attribute

async kill(restart: bool = False) → None: Kill the provisioner and optionally restart.

async launch_kernel(cmd: list[str], **kwargs: Any) → KernelConnectionInfo: Launch a kernel with a command.

async load_provisioner_info(provisioner_info: Dict) → None: Loads the base information necessary for persistence relative to this instance.

log: Logger or LoggerAdapter instance

notify_change(change): Notify observers of a change event

observe(handler: Callable[[...], Any], names: Union[traitlets.utils.sentinel.Sentinel, str, Iterable[traitlets.utils.sentinel.Sentinel | str]] = traitlets.All, type: traitlets.utils.sentinel.Sentinel | str = 'change') → None

Setup a handler to be called when a trait changes.

This is used to setup dynamic notifications of trait changes.

Parameters

handler (callable) – A callable that is called when a trait changes. Its signature should be handler(change), where change is a dictionary. The change dictionary at least holds a ‘type’ key. * type: the type of notification. Other keys may be passed depending on the value of ‘type’. In the case where type is ‘change’, we also have the following keys: * owner : the HasTraits instance * old : the old value of the modified trait attribute * new : the new value of the modified trait attribute * name : the name of the modified trait attribute.
names (list, str, All) – If names is All, the handler will apply to all traits. If a list of str, handler will apply to all names in the list. If a str, the handler will apply just to that name.
type (str, All (default: 'change')) – The type of notification to filter by. If equal to All, then all notifications are passed to the observe handler.

on_trait_change(handler=None, name=None, remove=False)

DEPRECATED: Setup a handler to be called when a trait changes.

This is used to setup dynamic notifications of trait changes.

Static handlers can be created by creating methods on a HasTraits subclass with the naming convention ‘_[traitname]_changed’. Thus, to create static handler for the trait ‘a’, create the method _a_changed(self, name, old, new) (fewer arguments can be used, see below).

If remove is True and handler is not specified, all change handlers for the specified name are uninstalled.

Parameters

handler (callable, None) – A callable that is called when a trait changes. Its signature can be handler(), handler(name), handler(name, new), handler(name, old, new), or handler(name, old, new, self).
name (list, str, None) – If None, the handler will apply to all traits. If a list of str, handler will apply to all names in the list. If a str, the handler will apply just to that name.
remove (bool) – If False (the default), then install the handler. If True then unintall it.

parent

A trait whose value must be an instance of a specified class.

The value can also be an instance of a subclass of the specified class.

Subclasses can declare default classes by overriding the klass attribute

pgid = None

pid = None

async poll() → Optional[int]: Poll the provisioner.

ports_cached = False

async post_launch(**kwargs: Any) → None

Perform any steps following the kernel process launch.

This method is called from KernelManager.post_start_kernel() as part of its start kernel sequence.

async pre_launch(**kwargs: Any) → Dict[str, Any]

Perform any steps in preparation for kernel process launch.

This includes applying additional substitutions to the kernel launch command and env. It also includes preparation of launch parameters.

Returns the updated kwargs.

process = None

classmethod section_names() → list[str]: return section names as a list

async send_signal(signum: int) → None

Sends a signal to the process group of the kernel (this usually includes the kernel and any subprocesses spawned by the kernel).

Note that since only SIGTERM is supported on Windows, we will check if the desired signal is for interrupt and apply the applicable code on Windows in that case.

set_trait(name, value): Forcibly sets trait attribute, including read-only attributes.

setup_instance(**kwargs): This is called before self.__init__ is called.

async shutdown_requested(restart: bool = False) → None

Allows the provisioner to determine if the kernel’s shutdown has been requested.

This method is called from KernelManager.request_shutdown() as part of its shutdown sequence.

This method is optional and is primarily used in scenarios where the provisioner may need to perform other operations in preparation for a kernel’s shutdown.

async terminate(restart: bool = False) → None: Terminate the provisioner and optionally restart.

trait_defaults(*names, **metadata)

Return a trait’s default value or a dictionary of them

Notes

Dynamically generated default values may depend on the current state of the object.

classmethod trait_events(name=None)

Get a dict of all the event handlers of this class.

Parameters: name (str (default: None)) – The name of a trait of this class. If name is None then all the event handlers of this class will be returned instead.
Return type: The event handlers associated with a trait name, or all event handlers.

trait_has_value(name)

Returns True if the specified trait has a value.

This will return false even if getattr would return a dynamically generated default value. These default values will be recognized as existing only after they have been generated.

Example

class MyClass(HasTraits):
    i = Int()

mc = MyClass()
assert not mc.trait_has_value("i")
mc.i # generates a default value
assert mc.trait_has_value("i")

trait_metadata(traitname, key, default=None): Get metadata values for trait by key.

trait_names(**metadata): Get a list of all the names of this class’ traits.

trait_values(**metadata)

A dict of trait names and their values.

The metadata kwargs allow functions to be passed in which filter traits based on metadata values. The functions should take a single value as an argument and return a boolean. If any function returns False, then the trait is not included in the output. If a metadata key doesn’t exist, None will be passed to the function.

Return type: A dict of trait names and their values.

Notes

Trait values are retrieved via getattr, any exceptions raised by traits or the operations they may trigger will result in the absence of a trait value in the result dict.

traits(**metadata)

Get a dict of all the traits of this class. The dictionary is keyed on the name and the values are the TraitType objects.

The TraitTypes returned don’t know anything about the values that the various HasTrait’s instances are holding.

The metadata kwargs allow functions to be passed in which filter traits based on metadata values. The functions should take a single value as an argument and return a boolean. If any function returns False, then the trait is not included in the output. If a metadata key doesn’t exist, None will be passed to the function.

unobserve(handler: Callable[[...], Any], names: Union[traitlets.utils.sentinel.Sentinel, str, Iterable[traitlets.utils.sentinel.Sentinel | str]] = traitlets.All, type: traitlets.utils.sentinel.Sentinel | str = 'change') → None

Remove a trait change handler.

This is used to unregister handlers to trait change notifications.

Parameters

handler (callable) – The callable called when a trait attribute changes.
names (list, str, All (default: All)) – The names of the traits for which the specified handler should be uninstalled. If names is All, the specified handler is uninstalled from the list of notifiers corresponding to all changes.
type (str or All (default: 'change')) – The type of notification to filter by. If All, the specified handler is uninstalled from the list of notifiers corresponding to all types.

unobserve_all(name: str | Any = traitlets.All) → None: Remove trait change handlers of any type for the specified name. If name is not specified, removes all trait notifiers.

update_config(config: traitlets.config.loader.Config) → None: Update config and load the new values

async wait() → Optional[int]: Wait for the provisioner process.

class euporie.core.kernel.MsgCallbacks

Bases: TypedDict

Typed dictionary for named message callbacks.

add_output: Callable[[dict[str, Any]], None] | None

ask_exit: Callable[[bool], None] | None

clear() → None. Remove all items from D.

clear_output: Callable[[bool], None] | None

completeness_status: Callable[[dict[str, Any]], None] | None

copy() → a shallow copy of D

dead: Callable[[], None] | None

done: Callable[[dict[str, Any]], None] | None

edit_magic: Callable[[str, int], None] | None

fromkeys(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() → a set-like object providing a view on D's items

keys() → a set-like object providing a view on D'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() → an object providing a view on D's values