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.AbstractEventLoop,

mod: types.ModuleType = ...,

modname: str = ...,

*,

context_factory: collections.abc.Callable[[], _contextvars.Context] = ...,

)

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

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

Ensures that the console is not left running after unset.

prehook(max_memerrs: int | None) → None

Ensures the console will be the only one running.

showtraceback() → None

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

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.AbstractEventLoop,

mod: types.ModuleType = ...,

modname: str = ...,

*,

context_factory: collections.abc.Callable[[], _contextvars.Context] = ...,

)

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.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: object,

) → None

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 used to substitute %-placeholders in template.

_interact_hook(ps1: object, kcolor: str, reset: str, fcolor: str) → None

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

In the main thread, the run method is preferred.

interrupt() → None

Pass additional_interrupt_hooks to the subclass constructor to change the behaviour when encountering a KeyboardInterrupt, instead of touching this method.

memoryerror() → None

Pass additional_memerr_hooks to the subclass constructor to change the behaviour when encountering a MemoryError, instead of touching this method.

posthook() → None

Called by run() after the interaction has ended before writing the exit message. Should not raise errors.

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

abstractmethod prehook(max_memerrs: int | None) → None

Called by run() before beginning the interaction logic. Can raise errors.

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

refresh() → None

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

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

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

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

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

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.

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, or 0 if the console has not exited.