sp

sp = subprocess wrappers

Classes:

• ProcessDt –

  Process time delta dataclass

• CompletedProcessDict –

  subprocess.CompletedProcess as a typed-dict

Functions:

• completed_process_dict –

  Convert CompletedProcess to CompletedProcessObj (typed dict)

• pcheck –

  Check process return code

• runb –

  Run a command capturing stdout/stderr as bytes

• runs –

  Run a command capturing stdout/stderr as strings

• run_dtee –

  Run a command, tee-ing stdout/stderr, and time it

• run_tee –

  Run a command, tee-ing stdout/stderr; see run_dtee

ProcessDt dataclass

ProcessDt(ti: float, tf: float, dt: float)

Process time delta dataclass

Examples:

>>> from shellfish.sp import ProcessDt
>>> ti = 0
>>> tf = 1
>>> dt = ProcessDt.from_titf(ti=ti, tf=tf)
>>> dt
ProcessDt(ti=0, tf=1, dt=1)

Methods:

• from_titf –

  Return a ProcessDt with dt computed from start/finish times

Attributes:

• ti (float) –

  Time the process started (seconds since epoch)

• tf (float) –

  Time the process finished (seconds since epoch)

• dt (float) –

  Time the process took to run (seconds; tf - ti)

ti instance-attribute

ti: float

Time the process started (seconds since epoch)

tf instance-attribute

tf: float

Time the process finished (seconds since epoch)

dt instance-attribute

dt: float

Time the process took to run (seconds; tf - ti)

from_titf classmethod

from_titf(ti: float, tf: float) -> ProcessDt

Return a ProcessDt with dt computed from start/finish times

Parameters:

• ti

  (float) –

  Time the process started (seconds since epoch)

• tf

  (float) –

  Time the process finished (seconds since epoch)

Returns:

• ProcessDt –

  ProcessDt object

CompletedProcessDict

Bases: TypedDict

subprocess.CompletedProcess as a typed-dict

Attributes:

• args (list[str]) –

  Command args the process was run with

• stdout (str) –

  Standard output (stdout) of the process

• stderr (str) –

  Standard error (stderr) of the process

• returncode (int) –

  Exit status of the process

args instance-attribute

args: list[str]

Command args the process was run with

stdout instance-attribute

stdout: str

Standard output (stdout) of the process

stderr instance-attribute

stderr: str

Standard error (stderr) of the process

returncode instance-attribute

returncode: int

Exit status of the process

completed_process_dict

completed_process_dict(
    completed_process: CompletedProcess[str],
) -> CompletedProcessDict

Convert CompletedProcess to CompletedProcessObj (typed dict)

Parameters:

• completed_process

  (CompletedProcess[str]) –

  CompletedProcess object

Returns:

• CompletedProcessDict –

  CompletedProcessObj typed dict

Examples:

>>> from subprocess import CompletedProcess
>>> from shellfish.sp import completed_process_dict
>>> cp = CompletedProcess(
...     args=['some', 'args'],
...     stdout="stdout string",
...     stderr="stderr string",
...     returncode=0
... )
>>> from pprint import pprint
>>> cp_typed_dict = completed_process_dict(completed_process=cp)
>>> pprint(cp_typed_dict)
{'args': ['some', 'args'],
 'returncode': 0,
 'stderr': 'stderr string',
 'stdout': 'stdout string'}

pcheck

pcheck(
    process: CompletedProcess[Any],
    ok_code: int
    | list[int]
    | tuple[int, ...]
    | set[int] = 0,
) -> None

Check process return code

Parameters:

• process

  (CompletedProcess[Any]) –

  CompletedProcess object

• ok_code

  (int | list[int] | tuple[int, ...] | set[int], default: 0 ) –

  OK code or sequence of codes, default is 0

Raises:

• CalledProcessError –

  if process.returncode is not ok

runb

runb(
    args: PopenArgs,
    *,
    executable: str | None = None,
    stdin: IO[Any] | int | None = None,
    input: str | None = None,
    stdout: IO[Any] | int | None = None,
    stderr: IO[Any] | int | None = None,
    shell: bool = False,
    cwd: FsPath | None = None,
    timeout: float | None = None,
    capture_output: bool = False,
    check: bool = False,
    env: Mapping[str, str] | None = None,
    ok_code: int
    | list[int]
    | tuple[int, ...]
    | set[int] = 0,
    **other_popen_kwargs: Any,
) -> CompletedProcess[bytes]

Run a command capturing stdout/stderr as bytes

Thin wrapper around subprocess.run with text=False and an ok_code aware check.

Parameters:

• args

  (PopenArgs) –

  Command args to run

• executable

  (str | None, default: None ) –

  Replacement program to execute

• stdin

  (IO[Any] | int | None, default: None ) –

  Stdin file/handle for the process

• input

  (str | None, default: None ) –

  Stdin to write to the process

• stdout

  (IO[Any] | int | None, default: None ) –

  Stdout file/handle for the process

• stderr

  (IO[Any] | int | None, default: None ) –

  Stderr file/handle for the process

• shell

  (bool, default: False ) –

  Run the command through the shell

• cwd

  (FsPath | None, default: None ) –

  Working directory to run the command in

• timeout

  (float | None, default: None ) –

  Timeout in seconds; None for no timeout

• capture_output

  (bool, default: False ) –

  Capture stdout and stderr

• check

  (bool, default: False ) –

  Check the return code against ok_code

• env

  (Mapping[str, str] | None, default: None ) –

  Environment variables to run the command with

• ok_code

  (int | list[int] | tuple[int, ...] | set[int], default: 0 ) –

  Return code (or collection of return codes) considered ok

• **other_popen_kwargs

  (Any, default: {} ) –

  Additional kwargs forwarded to subprocess.run

Returns:

• CompletedProcess[bytes] –

  subprocess.CompletedProcess with bytes stdout/stderr

Raises:

• CalledProcessError –

  If check and the return code is not ok

runs

runs(
    args: PopenArgs,
    *,
    executable: str | None = None,
    stdin: IO[Any] | int | None = None,
    input: str | None = None,
    stdout: IO[Any] | int | None = None,
    stderr: IO[Any] | int | None = None,
    shell: bool = False,
    cwd: FsPath | None = None,
    timeout: float | None = None,
    capture_output: bool = False,
    check: bool = False,
    env: Mapping[str, str] | None = None,
    ok_code: int
    | list[int]
    | tuple[int, ...]
    | set[int] = 0,
    **other_popen_kwargs: Any,
) -> CompletedProcess[str]

Run a command capturing stdout/stderr as strings

Thin wrapper around subprocess.run with text=True and an ok_code aware check.

Parameters:

• args

  (PopenArgs) –

  Command args to run

• executable

  (str | None, default: None ) –

  Replacement program to execute

• stdin

  (IO[Any] | int | None, default: None ) –

  Stdin file/handle for the process

• input

  (str | None, default: None ) –

  Stdin to write to the process

• stdout

  (IO[Any] | int | None, default: None ) –

  Stdout file/handle for the process

• stderr

  (IO[Any] | int | None, default: None ) –

  Stderr file/handle for the process

• shell

  (bool, default: False ) –

  Run the command through the shell

• cwd

  (FsPath | None, default: None ) –

  Working directory to run the command in

• timeout

  (float | None, default: None ) –

  Timeout in seconds; None for no timeout

• capture_output

  (bool, default: False ) –

  Capture stdout and stderr

• check

  (bool, default: False ) –

  Check the return code against ok_code

• env

  (Mapping[str, str] | None, default: None ) –

  Environment variables to run the command with

• ok_code

  (int | list[int] | tuple[int, ...] | set[int], default: 0 ) –

  Return code (or collection of return codes) considered ok

• **other_popen_kwargs

  (Any, default: {} ) –

  Additional kwargs forwarded to subprocess.run

Returns:

• CompletedProcess[str] –

  subprocess.CompletedProcess with str stdout/stderr

Raises:

• CalledProcessError –

  If check and the return code is not ok

run_dtee

run_dtee(
    args: PopenArgs,
    *,
    cwd: FsPath | None = None,
    env: dict[str, str] | None = None,
    input: STDIN | None = None,
    shell: bool = False,
    timeout: float | None = None,
) -> tuple[CompletedProcess[bytes], ProcessDt]

Run a command, tee-ing stdout/stderr, and time it

Streams the process' stdout/stderr to this process' stdout/stderr as they arrive, while also capturing them.

Parameters:

• args

  (PopenArgs) –

  Command args to run

• cwd

  (FsPath | None, default: None ) –

  Working directory to run the command in

• env

  (dict[str, str] | None, default: None ) –

  Environment variables to run the command with

• input

  (STDIN | None, default: None ) –

  Stdin to write to the process

• shell

  (bool, default: False ) –

  Run the command through the shell

• timeout

  (float | None, default: None ) –

  Timeout in seconds; None for no timeout

Returns:

• tuple[CompletedProcess[bytes], ProcessDt] –

  Tuple of the subprocess.CompletedProcess and its ProcessDt timing

Raises:

• TimeoutExpired –

  If the process does not finish within timeout

run_tee

run_tee(
    args: PopenArgs,
    *,
    cwd: str | None = None,
    env: dict[str, str] | None = None,
    input: STDIN | None = None,
    shell: bool = False,
    timeout: float | None = None,
) -> CompletedProcess[bytes]

Run a command, tee-ing stdout/stderr; see run_dtee

Parameters:

• args

  (PopenArgs) –

  Command args to run

• cwd

  (str | None, default: None ) –

  Working directory to run the command in

• env

  (dict[str, str] | None, default: None ) –

  Environment variables to run the command with

• input

  (STDIN | None, default: None ) –

  Stdin to write to the process

• shell

  (bool, default: False ) –

  Run the command through the shell

• timeout

  (float | None, default: None ) –

  Timeout in seconds; None for no timeout

Returns:

• CompletedProcess[bytes] –

  subprocess.CompletedProcess with bytes stdout/stderr