vcorelib.asyncio package

Submodules

vcorelib.asyncio.cli module

A module implementing utilities to work with asyncio processes.

class vcorelib.asyncio.cli.ProcessResult(proc: Process, stdout: bytes | None = None, stderr: bytes | None = None)

Bases: NamedTuple

A process result (after calling ‘communicate’).

proc: Process

Alias for field number 0

stderr: bytes | None

Alias for field number 2

stdout: bytes | None

Alias for field number 1

async vcorelib.asyncio.cli.create_subprocess_shell_log(logger: Logger | LoggerAdapter[Any], command: str, stdout: int = None, stderr: int = None, **kwargs) → Process

Create a shell process and log information about the created process.

async vcorelib.asyncio.cli.handle_process_cancel(proc: Process, name: str, logger: Logger | LoggerAdapter[Any], stdin: bytes = None, signal: int = None) → ProcessResult

Communicate with a process and send a signal to it if this task gets cancelled.

async vcorelib.asyncio.cli.run_command(logger: Logger | LoggerAdapter[Any], *args: str, stdin: bytes = None, stdout: int = None, stderr: int = None, signal: int = None, **kwargs) → ProcessResult

Run a subprocess and return the completed process.

async vcorelib.asyncio.cli.run_shell(logger: Logger | LoggerAdapter[Any], *args: str, stdin: bytes = None, stdout: int = None, stderr: int = None, signal: int = None, **kwargs) → ProcessResult

Run a shell command and return the completed process.

vcorelib.asyncio.subprocess module

A module implementing utilities for working with asyncio subprocesses.

async vcorelib.asyncio.subprocess.create_subprocess_exec_log(logger: Logger | LoggerAdapter[Any], program: str, *args: str, stdout: int = None, stderr: int = None, rel: Tuple[str, str] = None, **kwargs) → Process

Create a subprocess and log information about the created process. Reconcile subtle platform differences in running simple commands.

vcorelib.asyncio.subprocess.log_process_info(program: str, *args: str, base: Path | str | None = None) → Tuple[str, str]

Get a relative-path program string and a space-delimeted string of the arguments, which have also been shortened against a possible relative path.

Module contents

A module for working with asyncio.

vcorelib.asyncio.all_stop_signals() → Set[int]

Get a set of all stop signals on this platform.

vcorelib.asyncio.event_setter(stop_sig: Event, eloop: AbstractEventLoop = None, logger: Logger | LoggerAdapter[Any] = None) → Callable[[int, FrameType | None], None]

Create a function that sets an event.

vcorelib.asyncio.log_exceptions(tasks: Iterable[Task[T]], logger: Logger | LoggerAdapter[Any] = None) → List[Task[T]]

Log task exception and return the list of tasks that aren’t complete.

vcorelib.asyncio.log_task_exception(task: Task[Any], logger: Logger | LoggerAdapter[Any] = None) → None

If a task is done and raised an exception, log it.

vcorelib.asyncio.new_eloop(set_current: bool = True) → AbstractEventLoop

Get a new event loop.

vcorelib.asyncio.normalize_eloop(eloop: AbstractEventLoop = None) → AbstractEventLoop

Get the active event loop if one isn’t provided explicitly.

vcorelib.asyncio.run_handle_interrupt(to_run: Awaitable[T], eloop: AbstractEventLoop = None, enable_uvloop: bool = True) → T | None

Run a task in an event loop and gracefully handle keyboard interrupts.

Return the result of the awaitable or None if execution was interrupted.

vcorelib.asyncio.run_handle_stop(stop_sig: Event, task: Coroutine[None, None, T], eloop: AbstractEventLoop = None, signals: Iterable[int] = None, enable_uvloop: bool = True) → T

Publish the stop signal on keyboard interrupt and wait for the task to complete.

vcorelib.asyncio.shutdown_loop(eloop: AbstractEventLoop = None, logger: Logger | LoggerAdapter[Any] = None) → None

Attempt to shut down an event loop.

vcorelib.asyncio.try_uvloop_runner(debug: bool = None, eloop: AbstractEventLoop = None, enable: bool = True) → Iterator[AbstractEventLoop]

Try to set up an asyncio runner using uvloop.