process module

Classes for handling OS processes and background tasks.

They integrate low-level Python constructs like subprocess.Popen and asyncio.create_subprocess_exec() into the ctlbase package.

Classes

class ctlbase.process.TaskMonitor(loop, taskCheckingInterval_ms=1000, msgBuffer=None, ns=True)

A monitor for asyncio tasks.

It awaits tasks running in the background, avoiding warnings like “Task was destroyed but it is pending!” or “coroutine was never awaited”. What is more, results and error messages from completed tasks can be appended to a message buffer.

Tasks to be monitored are added via add(). They are regularly checked for success or failure.

__init__(loop, taskCheckingInterval_ms=1000, msgBuffer=None, ns=True)

Creates a task monitor.

Parameters:

  • loop (asyncio.EventLoop) – See loop.

  • taskCheckingInterval_ms (int) – See taskCheckingInterval_ms.

  • msgBuffer (MessageBuffer | None) – The message buffer to which results and error messages of completed tasks are appended. If None, results and errors are silently ignored.

  • ns (str) – The message namespace for appended messages. If True, the namespace is derived from the caller name. See CallerInspector for the discovery algorithm. Only relevant if msgBuffer is not None.

add(task, name=None)

Adds a task to the task monitor.

Parameters:

task (coroutine | asyncio.Task) – The task to be added. If it is a coroutine, it is scheduled as a task in the event loop.

Raises:

asyncio.InvalidStateError – If the event loop is already closed.

Returns:

The added task. Only relevant if task is a coroutine.

Return type:

asyncio.Task

async cancelTasks()

Cancels all tasks in the task monitor and waits for them to complete.

cancelTasksInLoop()

Cancels all tasks in the event loop and waits for them to complete.

This applies to all the tasks in the event loop, no matter if they were added to the task monitor ot not.

remove(task)

Removes a task from the task monitor without cancelling it.

Parameters:

task (asyncio.Task) – The task to be removed.

start()

Creates and starts a monitoring thread in the event loop.

Tasks can be added at any time, that is, before and after the monitor is running.

Added tasks are checked every taskCheckingInterval_ms milliseconds for their state. If a task completed successfully, its result is appended to the message buffer with key TASK_SUCCEEDED unless it is None. If a task failed, the error message of its exception is appended to the message buffer with key TASK_FAILED. In any case, a completed task is removed from the task monitor.

async stop()

Stops the task monitor and cancels all its tasks.

loop: asyncio.EventLoop

The asyncio event loop in which new tasks are created, including the monitoring thread itself.

taskCheckingInterval_ms: int

The interval in milliseconds in which tasks are checked for their state.

class ctlbase.process.CommandExecutor(msgBuffer=None, ns=True, isSynchronousDefault=False)

An executor for commands, i.e. executables with optional arguments. Each command is executed in a subprocess.

Supported variants:

  • synchronous or asynchronous execution

  • a single command or multiple commands

  • as an executable with distinct arguments or as a shell command

While the class follows an asyncio design, commands may or may not be executed in an asyncio event loop.

exception SynchronousError(returncode=None, message=None)

An error that occurred while synchronously executing a command.

lng: str | None

A two-letter language code for the language of the message text.

message: str

The message text of the error returned by the command.

returncode: int

The numeric return code of the command.

__init__(msgBuffer=None, ns=True, isSynchronousDefault=False)

Creates a new command executor.

Parameters:

  • msgBuffer (LoggingMessageBuffer) – The message buffer to which messages are logged. Logging is always asynchronous. The actual log output can be controlled per execution via the isLogEnabled parameter of the execution method.

  • ns (str) – The message namespace for logged messages. If True, the namespace is derived from the caller name. See CallerInspector for the discovery algorithm.

  • isSynchronousDefault (bool) – See isSynchronousDefault.

async _createProcess(command, env=None, isStdIn=False, isStdOut=True, isStdErr=True, useAsyncio=False)

Creates a subprocess for executing a command.

The command can be given in one of two forms:

  1. As a list or tuple with one or more str instances. The first one denotes the executable, possibly with a path specification. All remaining str instances denote the arguments to be passed to the executable.

  2. As a single str denoting a shell command.

If form 1 is used without a path specification, the path to the executable is resolved via findBinary() and shutil.which(), then cached to speed up subsequent calls.

Parameters:

  • command (list[str] | tuple[str] | str) – The command to execute, depending on the chosen form (see above).

  • env (dict) – Environment variables to be passed to the subprocess, as defined for subprocess.Popen.

  • isStdIn (bool) – If True, the subprocess is prepared for receiving data from the parent process via stdin.

  • isStdOut (bool) – If True, the parent process is prepared for receiving data from the subprocess via stdout.

  • isStdErr (bool) – If True, the parent process is prepared for receiving data from the subprocess via stderr.

  • useAsyncio – If True, the subprocess is created with asyncio. Consequently, the communication with the subprocess takes place in an asyncio event loop.

Raises:

FileNotFoundError – If the executable could not be resolved to a path.

Returns:

The created subprocess. Its type depends on the value of useAsyncio.

Return type:

subprocess.Popen | asyncio.subprocess.Process

async executeAsynchronously(command, env=None, isLogEnabled=True)

Executes a single command in a subprocess without waiting for completion.

Parameters:
Raises:

FileNotFoundError – If the executable could not be resolved to a path.

async executeCommands(commands, stdInput=None, isSynchronous=None, timeout_ms=None, outputEncoding='utf-8', isLogEnabled=True)

Executes one or more commands with the given options.

A single command can be given in one of the forms defined for _createProcess().

What is more, if command form 1 is used, multiple commands can be wrapped in another list or tuple.

Parameters:
Raises:
Returns:

A list of bytes if isSynchronous evaluates to True and outputEncoding is None. Each entry represents the data received from one subprocess via stdout. Index 0 corresponds to the first executed command, i.e. index 0 in commands. Index 1 corresponds to the second command and so on.
If outputEncoding is not None all stdout output is concatenated to one str with newlines.
None if isSynchronous evaluates to False.

Return type:

list[bytes] | str | None

async executeSynchronously(command, env=None, stdInput=None, timeout_ms=None, outputEncoding='utf-8', isLogEnabled=True)

Executes a single command in a subprocess and waits for its completion in an asyncio event loop.

Parameters:

  • command (list[str] | tuple[str] | str) – The command to execute in one of the forms defined for _createProcess().

  • env (dict) – Environment variables to be passed to the subprocess, as defined for subprocess.Popen.

  • stdInput (bytes | bytearray | str | None) – Data to be sent to the subprocess via stdin. If None, no data is sent.

  • timeout_ms (int | None) – Timeout in milliseconds to wait for the subprocess to complete. If the timeout expires, the subprocess is killed. If None, the subprocess might run indefinitely.

  • outputEncoding (str | None) – An encoding specifier like utf-8 for decoding the bytes from stdout and stderr into a str. If None, bytes are returned unmodified.

  • isLogEnabled (bool | str) – If True, all occurring messages are written to the command executor’s message buffer. For more fine-grained behavior, use the values ON_SUCCESS_ONLY and ON_ERROR_ONLY.

Raises:
Returns:

Data received from the subprocess via stdout, possibly decoded to a str. See outputEncoding for decoding options.

Return type:

bytes | str

async getOutput(command)

Executes a single command synchronously and returns the output of stdout or stderr as an UTF-8 str.

Hence, it is a simplified version of executeSynchronously() for retrieving the output of an command. No logging is done.

Possibly existing ANSI control characters are removed from the output, so it can be easily processed programmatically.

Raises:

FileNotFoundError – If the executable could not be resolved to a path.

Returns:

The output of stdout if the exit code of the subprocess is 0. The output of stderr otherwise.

Return type:

str

ON_ERROR_ONLY = 'ON_ERROR_ONLY'

If passed to isLogEnabled, only error messages are logged.

ON_SUCCESS_ONLY = 'ON_SUCCESS_ONLY'

If passed to isLogEnabled, only success messages are logged.

isSynchronousDefault: bool

If True commands are executed synchronously by default.

Constants and defaults

ctlbase.process.TASK_CHECKING_INTERVAL_MS_DEFAULT = 1000

Default interval in milliseconds in which tasks are checked for their state.