asyncutils.console

asyncutils.console#

Implementation of an interactive async console base class, as well as an AsyncUtilsConsole class derived from it.

Classes#

`AsyncUtilsConsole`	A subclass of `ConsoleBase`, used to implement the `asyncutils` REPL.
`ConsoleBase`	A base class for async consoles. Derives from `InteractiveConsole`, or `_pyrepl.console.InteractiveColoredConsole` if available. It is inspired by asyncio and highly adaptable.

Module Contents#

class asyncutils.console.AsyncUtilsConsole( loop: asyncio.AbstractEventLoop, mod: types.ModuleType = ..., modname: str = ..., *, context_factory: collections.abc.Callable[[], contextvars.Context] = ..., )[source]#

Bases: ConsoleBase

A subclass of ConsoleBase, used to implement the asyncutils REPL.

loop (required): Event loop used by console interaction.
mod: The module to import within the console.
modname: The name of the above module, determined by the subclass name by default.
context_factory: A function that takes no arguments and returns an instance of contextvars.Context, to be used by the event loop.

_interact_hook(ps1: object, kcolour: str, reset: str, fcolour: str) → None[source]#: Write code with emulated colour (such as import statements to represent the namespace) after the banner has been written, with parameters ps1 representing sys.ps1 and kcolour, reset and fcolour representing the ANSI escape codes for the keyword colour, colour reset and the function colour respectively.

after_run() → None[source]#: Ensure the console is not left running after unset.

before_run(max_memory_errors: int | None) → None[source]#: Ensure the console will be the only one running.

showtraceback() → None[source]#: Display the formatted traceback of the exception being handled. If there was no exception, do nothing.

Note

This differs from the superclass behaviour, where AttributeError is raised outright.

write_special(msg: str) → None[source]#: Write msg to stderr iff the quiet flag is not set.

property is_running: bool#: Whether the console is currently running. Also performs internal state consistency checks, asserting that only one AsyncUtilsConsole can be running at a time.

class asyncutils.console.ConsoleBase( loop: asyncio.AbstractEventLoop, mod: types.ModuleType = ..., modname: str = ..., *, context_factory: collections.abc.Callable[[], contextvars.Context] = ..., )[source]#

Bases: code.InteractiveConsole, abc.ABC

A base class for async consoles. Derives from InteractiveConsole, or _pyrepl.console.InteractiveColoredConsole if available. It is inspired by asyncio and highly adaptable.

loop (required): Event loop used by console interaction.
mod: The module to import within the console.
modname: The name of the above module, determined by the subclass name by default.
context_factory: A function that takes no arguments and returns an instance of contextvars.Context, to be used by the event loop.

classmethod __init_subclass__(

*,

name: str = ...,

version: str = ...,

description: str = ...,

default_local_exit: bool = ...,

disallow_subclass_msg: str | None = ...,

native_handler: collections.abc.Callable[[dict[str, Any]], object] | None = ...,

other_handlers: dict[str, collections.abc.Callable[[dict[str, Any]], object] | None] = ...,

additional_interrupt_hooks: collections.abc.Iterable[collections.abc.Callable[[Self], object]] = ...,

additional_memory_error_hooks: collections.abc.Iterable[collections.abc.Callable[[Self], object]] = ...,

template: str = ...,

**k: object,

) → None[source]#

All of the arguments below are optional.

name: name of the module using the console
version: version of the module using the console
description: description of the module using the console
default_local_exit, disallow_subclass_msg, native_handler, other_handlers, additional_interrupt_hooks, additional_memory_error_hooks: see above
template: the console banner to use, with %-placeholders for name, version and description

Additional keyword arguments are used to substitute %-placeholders in ``template``.

_interact_hook(ps1: object, kcolour: str, reset: str, fcolour: str) → None[source]#: Write code with emulated colour (such as import statements to represent the namespace) after the banner has been written, with parameters ps1 representing sys.ps1 and kcolour, reset and fcolour representing the ANSI escape codes for the keyword colour, colour reset and the function colour respectively.

after_run() → None[source]#: Finalize the interaction from run(). Called before writing the exit message. Should not raise errors.

It is highly recommended that subclasses implement this and call super().after_run() within the implementation after the custom logic.

abstractmethod before_run(max_memory_errors: int | None) → None[source]#: Prepare for the interaction logic to begin in run(). Can raise errors.

When implementing, call super().before_run(max_memory_errors) before everything. This allows subclasses to pass their own value of max_memory_errors.

interact(banner: str | None = ..., *, ps1: object = ...) → None[source]#: In the main thread, the run method is preferred.

interrupt() → None[source]#: Pass additional_interrupt_hooks to the subclass constructor to change the behaviour when encountering a KeyboardInterrupt, instead of touching this method.

memory_error() → None[source]#: Pass additional_memory_error_hooks to the subclass constructor to change the behaviour when encountering a MemoryError, instead of touching this method.

refresh() → None[source]#: Cancel the internal future such that it no longer resolves to the result of an async call. Called by interrupt() and memory_error().

run( *, exit_message: str = ..., thread_name: str = ..., max_memory_errors: int = ..., always_run_interactive: bool = ..., always_install_completer: bool = ..., suppress_asyncio_warnings: bool = ..., suppress_unawaited_coroutine_warnings: bool = ..., ) → int[source]#: Run the console and return the integer return code.

The strings exit_message and thread_name should support %-formatting, the placeholder being the module name.

Pass a negative value for max_memory_errors to disable the stop after certain number of MemoryError’s behaviour.

If always_install_completer is True, set the completer on readline as long as readline is available.

Pass True for suppress_asyncio_warnings and suppress_unawaited_coroutine_warnings to silence asyncio logging and warnings for garbage-collected coroutines not being awaited respectively.

If you wish the console to act like a console even when stdin is piped, pass always_run_interactive=True or start Python with the -i flag.

runcode( code: types.CodeType, *, fimp: collections.abc.Callable[[], concurrent.futures.Future[Any]] = ..., no_traceback: tuple[asyncutils._internal.prots.ExcType, Ellipsis] = ..., threadsafe: bool = ..., ) → Any | None[source]#: Run code, an instance of types.CodeType, as a callback managed by the event loop, and return its result, or STATEMENT_FAILED if the statement fails.

fimp is a function that returns an instance of concurrent.futures.Future.

no_traceback is a tuple of types of exceptions for which the traceback should not be shown if they are to occur.

threadsafe dictates whether to run the code in the event loop using call_soon_threadsafe() instead of call_soon().

set_return_code(exc: SystemExit, /) → None[source]#
set_return_code(code: int | str, /) → None: Set the return code of this console from an instance of SystemExit or an integer return code and exit the console.

showtraceback() → None[source]#: Display the formatted traceback of the exception being handled. If there was no exception, do nothing.

Note

This differs from the superclass behaviour, where AttributeError is raised outright.

write_special(msg: str) → None[source]#: Write the banner and exit messages. Can have a different implementation than write().

BANNER: ClassVar[str]#: A %-formattable string representing the template of the banner to be shown when the console starts.

CAN_USE_PYREPL: ClassVar[bool]#: Whether _pyrepl enhancements are available and allowed.

LOCALS_HANDLERS: ClassVar[collections.ChainMap[str, collections.abc.Callable[[dict[str, Any]], Any] | None]]#: Maps module names to a function taking locals of a console of the corresponding type. The return value is discarded.

Add handlers for the module of your own console with native_handler and other modules with other_handlers.

NAME: ClassVar[str]#: The name of the module implementing this console, detected from the class name if the keyword argument name is not provided to the subclass constructor.

property _internal_is_running: bool#: Whether the console thinks itself is running. Can be used in is_running for state consistency checks.

property context: contextvars.Context#: The contextvars.Context instance passed to methods of the underlying asyncio event loop.

default_local_exit: ClassVar[bool]#: Whether Python should continue running after the console exits by default, as opposed to the console raising SystemExit directly.

disallow_subclass_msg: ClassVar[str]#: The error message when attempts are made to subclass subclasses of this class. Specified through the disallow_subclass_msg argument, which any unsubclassable console should pass.

property exc: SystemExit | None#: The SystemExit instance that caused the console to exit, or None if the console has not exited.

interrupt_hooks: ClassVar[tuple[collections.abc.Callable[[Self], Any], Ellipsis]]#: Functions called when KeyboardInterrupt occurs, in that order, besides essential hardcoded logic.

Note

Add hooks using the additional_interrupt_hooks class construction parameter.

property is_running: bool#: Whether the console is running. The default implementation simply returns _internal_is_running.

memory_error_hooks: ClassVar[tuple[collections.abc.Callable[[Self], Any], Ellipsis]]#: Functions called when a MemoryError occurs, in that order, besides essential hardcoded logic.

Note

Add hooks using the additional_memory_error_hooks class construction parameter.

property memory_errors: int#: The number of MemoryError’s that have occurred.