pyantz.jobs.subproc package

pyantz.jobs.subproc package#

Submodules#

pyantz.jobs.subproc.dispatch module#

Run arbitrary executables as subprocesses.

dispatch() wraps Python’s subprocess.run() with optional stdout/stderr capture to files, environment variable overrides, a working directory, and a timeout. It returns True on success (exit code 0) and False on subprocess.CalledProcessError or timeout.

Bases: BaseModel

Parameters for dispatching jobs.

Parameters:

cmd (tuple[str, ...])
timeout (float | None)
input_stream (bytes | None)
cwd (Annotated[Path, PathType(path_type=dir)] | None)
env (Mapping[str, str] | None)
stdout_file (Path | str | None)
stderr_file (Path | str | None)

_abc_impl = <_abc._abc_data object>#

cmd: tuple[str, ...]#

the program followed by its arguments.

Type:: Command to run

cwd: Path | None#: Working directory of the subprocess; it must exist.

env: Mapping[str, str] | None#: If not None, the environment variables of the new process, used instead of inheriting the current process environment.

input_stream: bytes | None#: Input to provide to the subprocess in stdin.

stderr_file: Path | str | None#: File to write stderr to. If the same as stdout_file, stderr is appended after stdout and a header.

stdout_file: Path | str | None#: File to write stdout to, overwriting any existing file. If the same as stderr_file, it starts with stdout and then has stderr appended after a header.

timeout: float | None#: If set, and if the timeout (in seconds) expires, the child process is killed and waited for.

STDERR_HEADER = b'\n--- stderr ---\n'#: Separates stdout from stderr when both are written to the same file.

_save_output(params: DispatchParams, stdout: bytes, stderr: bytes)[source]#

Write the captured output to the files the user asked for.

Missing parent directories are created. When stdout_file and stderr_file are the same file, it holds stdout, STDERR_HEADER and then stderr.

Parameters:

params (DispatchParams) – the dispatch parameters naming the output files.
stdout (bytes) – the captured stdout.
stderr (bytes) – the captured stderr.

Return type:

None

Returns:

None.

Raises:

OSError – if a file can’t be written.

Example

from pathlib import Path

from pyantz.jobs.subproc.dispatch import DispatchParams, _save_output

params = DispatchParams(
    cmd=("unused",), stdout_file="logs/out.txt", stderr_file="logs/out.txt"
)
_save_output(params, b"some output", b"a warning")
print(Path("logs/out.txt").read_text())

some output
--- stderr ---
a warning

dispatch(params: DispatchParams)[source]#

Dispatch a command in a subprocess.

The output is saved to the requested files whether or not the command succeeds. The command is run directly, without a shell.

Parameters:

params (DispatchParams) – the job’s parameters, validated as DispatchParams.

Job parameters:

cmd (tuple[str, …]) – the program to run followed by its arguments.
timeout (float | None) – seconds after which the process is killed. Defaults to None (no timeout).
input_stream (bytes | None) – input for the process’s stdin. Defaults to None.
cwd (Path | None) – working directory of the process, which must exist. Defaults to None (this process’s).
env (Mapping[str, str] | None) – environment variables of the process, replacing this process’s. Defaults to None (inherit them).
stdout_file (Path | str | None) – file to write stdout to. Defaults to None.
stderr_file (Path | str | None) – file to write stderr to; it may be stdout_file. Defaults to None.

Returns:

True if the command exited with code 0, False if it exited with another code, timed out or couldn’t be started.

Return type:

bool

Raises:

OSError – if an output file can’t be written.

Example

import sys
from pathlib import Path

from pyantz.jobs.subproc.dispatch import dispatch

success = dispatch(
    {
        "cmd": [sys.executable, "-c", "print(input().upper())"],
        "input_stream": b"hello",
        "stdout_file": "stdout.txt",
    },
    lambda job: None,
)
print(success, Path("stdout.txt").read_text().strip())

True HELLO

Module contents#

Jobs to run other processes or files.

pyantz.jobs.subproc package

Contents

pyantz.jobs.subproc package#

Submodules#

pyantz.jobs.subproc.dispatch module#

Module contents#