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 `InteractiveColoredConsole` if available. It is inspired by `asyncio.__main__` and highly adaptable.

Module Contents#

class asyncutils.console.AsyncUtilsConsole(loop: asyncio.events.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, determined by the subclass name by default.
modname: The name of the above module.
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, kcolor: str, reset: str, fcolor: str) → None[source]#: Called to write code with emulated color (such as import statements to represent the namespace) after the banner has been written, with parameters ps1 representing sys.ps1 and kcolor, reset and fcolor representing the ANSI escape codes for the keyword color, color reset and the function color respectively.

posthook() → None[source]#: Ensures that the console is not left running after unset.

prehook(max_memerrs: int) → None[source]#: Ensures 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 (this differs from the superclass behaviour).

write_special(msg: str) → None[source]#: Writes msg to stderr if and only if the quiet flag is not set.

property is_running: bool#: Performs internal state consistency checks and returns whether the console is currently running. Only one AsyncUtilsConsole can be running at a time.

class asyncutils.console.ConsoleBase(loop: asyncio.events.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 InteractiveColoredConsole if available. It is inspired by asyncio.__main__ and highly adaptable.

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

__callback(fut: concurrent.futures.Future[Any], code: types.CodeType, /, *, makef: _collections_abc.Callable[[types.CodeType, dict[str, Any]], _collections_abc.Callable[[], Any]] = ..., corocheck: _collections_abc.Callable[[object], TypeGuard[_collections_abc.Coroutine[Any, Any, Any]]] = ..., futchain: _collections_abc.Callable[[asyncio.tasks.Task[Any], concurrent.futures.Future[Any]], object] = ...) → None#: Called by runcode internally. To change its behaviour, override the entire method in a subclass with different default parameters.

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_memerr_hooks: _collections_abc.Iterable[_collections_abc.Callable[[Self], object]] = ..., template: str = ..., **k: Any) → 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_memerr_hooks: see above

template: the console banner to use, with %-placeholders for name, version and description

Additional keyword arguments are passed to template.__mod__().

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

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.

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

abstractmethod posthook() → None[source]#: Called by run() after the interaction has ended before writing the exit message. Should not raise errors.

When implementing, call super().posthook() after everything.

Not really an abstract method, but implementing is highly recommended.

abstractmethod prehook(max_memerrs: int) → None[source]#: Called by run() before beginning the interaction logic. Can raise errors.

When implementing, call super().prehook(max_memerrs) before everything. This allows subclasses to provide their own value of max_memerrs and change the signature of the prehook.

Not really an abstract method, but implementing is highly recommended.

refresh() → None[source]#: Callback in interrupt() and memoryerror().

run(*, exitmsg: str = ..., threadname: str = ..., max_memerrs: 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 exitmsg and threadname should support %-formatting, the placeholder being the module name.

Pass a negative value for max_memerrs 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, *, futimpl: _collections_abc.Callable[[], concurrent.futures.Future[Any]] = ..., dont_show_traceback: tuple[asyncutils._internal.types.ExcType, Ellipsis] = ..., threadsafe: bool = ...) → Any | None[source]#: Run code, an instance of types.CodeType.

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

dont_show_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 (this differs from the superclass behaviour).

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

BANNER: ClassVar[str]#: A %-formattable string representating 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]]#: module name -> (locals of console of corresponding type -> Any)

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.

STATEMENT_FAILED: ClassVar[object]#: This is present if and only if _pyrepl.console.InteractiveColoredConsole is used as the parent of this class.

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.

memerr_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_memerr_hooks class construction parameter.

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

property retcode: int#: The integer return code of the console. If the console has not exited, return 0.