Commands and results · API reference

module

`sh`

from cuprum import sh

Safe command construction and execution facade for curated programs.

This package focuses on the typed core: building SafeCmd instances from curated Program values and providing a minimal async runtime for executing them with predictable semantics.

The implementation is split across submodules by responsibility:

- argv builds argument vectors and publishes ArgValue. - builder holds the SafeCmdBuilder callable contract. - execution holds the execution context, stdin, and timeout types. - results holds CommandResult, PipelineResult, and their byte-exact counterparts BytesCommandResult and BytesPipelineResult. - output holds RunOutputOptions and IOOptions. - stdio holds the StdioTarget standard-stream binding vocabulary. - safe_cmd holds the SafeCmd execution primitive. - pipeline holds the Pipeline composition primitive. - factory holds the make builder factory.

Every name previously defined or imported by the former cuprum/sh.py module remains importable from cuprum.sh with the same object identity; the package is the stable public and internal entry point.

2 exports

function sh.build_argv

def build_argv(*args: ArgValue, **kwargs: ArgValue) -> tuple[str, ...]

Build an argv tuple using the same rules as sh.make builders.

Parameters

*args: Positional text, integer, float, boolean, or path values. They are stringified in the order supplied and appear before generated keyword flags.
**kwargs: Keyword text, integer, float, boolean, or path values. Each key is normalized by replacing underscores with hyphens, then serialized as --flag=value in insertion order.

Returns

tuple[str, ...]: The constructed argv tuple, excluding the program name.

Raises

TypeError: If any value is None, is a bytes value, or is not a str, int, float, bool, or path-like value, in either position. A path-like whose __fspath__ returns bytes is rejected as bytes. Rejected bytes values are quoted in the message.

Examples

>>> build_argv("status", porcelain=True, branch="main")
('status', '--porcelain=True', '--branch=main')

function sh.make

def make(
    program: Program,
    *,
    catalogue: ProgramCatalogue | None = None,
) -> SafeCmdBuilder

Build a callable that produces SafeCmd instances for program.

The catalogue is resolved once, when this function is called, in this order:

1. an explicit catalogue argument; 2. the innermost active scoped(catalogue=...); 3. DEFAULT_CATALOGUE, outside any catalogue scope.

The resolved catalogue is bound to the returned builder, so the builder keeps working after the scope that supplied it exits. Passing a scope with only an allowlist (ScopeConfig(allowlist=...)) does not change which catalogue is active.

Membership is checked here, at construction. Permission is a separate check: the allowlist of whichever context is active when the command runs decides whether it may execute. A command built from a catalogue the enclosing scope does not allow therefore constructs successfully and raises ForbiddenProgramError at run or run_sync time.

Parameters

program Program: The program the built SafeCmd instances invoke; it must exist in the resolved catalogue.
catalogue ProgramCatalogue | None, optional: The catalogue used to validate program and resolve its entry. Defaults to the innermost scoped catalogue, then DEFAULT_CATALOGUE.

Returns

SafeCmdBuilder: A callable that builds SafeCmd instances for program. It accepts the same positional and keyword values the runtime validates: str, int, float, bool, or a text path-like value.

Raises

UnknownProgramError: If program does not exist in the resolved catalogue.

class

`SafeCmd`

cuprum/sh/safe_cmd.py

class SafeCmd(
    program: Program,
    argv: tuple[str, ...],
    project: ProjectSettings,
    __weakref__: object = dc.field(init=False, repr=False, hash=False, compare=False),
)

Bases: _OwnedStart, _BytesEntryPoints

Typed representation of a curated command ready for execution.

10 members

field SafeCmd.program

program: Program

field SafeCmd.argv

argv: tuple[str, ...]

field SafeCmd.project

project: ProjectSettings

property SafeCmd.argv_with_program

argv_with_program: tuple[str, ...]

The program name followed by this command's arguments.

Returns

tuple[str, ...]: An argument vector whose first item is str(program).

method SafeCmd.run

async def run(
    *,
    output: RunOutputOptions | None = None,
    timeout: float | None = None,
    context: ExecutionContext | None = None,
    stdin: StdinSource | None = None,
) -> CommandResult

Execute the command asynchronously with predictable cancellation.

Parameters

output RunOutputOptions | None, default=None: Capture and echo settings. Its 64 KiB default bounds each mirrored line without affecting capture; set max_echo_line_bytes=None for unbounded mirroring.
timeout float | None, default=None: Maximum execution time in seconds. An explicit value overrides the timeout in context.
context ExecutionContext | None, default=None: Execution settings, including echo sinks and text encoding.
stdin StdinInput | StdinStream | None, default=None: Optional stdin source. StdinInput supplies one complete bytes or text payload; StdinStream supplies an async producer whose chunks are pulled one at a time, written, and drained before the next is pulled. That backpressure limits how far *ahead* the producer runs — it does not stop the writer pulling ahead of the child's reads, because drain() returns at the transport's low-water mark rather than on an empty buffer, and the OS pipe holds bytes of its own. The payload is never held whole, and retained memory is roughly the chunk currently being written plus those buffers, so yield bounded-size chunks when that matters. None inherits the parent's stdin.

Returns

CommandResult: The command outcome, including complete captured streams when output.capture is true.

Raises

PermissionError: If the command is not allowed by the active scope.
TimeoutError: If execution exceeds the effective timeout.
UnicodeEncodeError: If text stdin cannot be encoded by the execution context.
StdinSourceError: If a StdinStream producer or its encoder fails. The child is terminated before this is raised.

method SafeCmd.lines

def lines(
    *,
    output: RunOutputOptions | None = None,
    timeout: float | None = None,
    context: ExecutionContext | None = None,
    stdin: StdinSource | None = None,
) -> LineStream

Iterate the command's output lines as they arrive.

Line events are delivered in arrival order per stream, stamped with monotonic seconds since the command started. Capture and echo stay governed by *output* independently: iterating lines does not disable either unless the caller asks.

Parameters

output: Optional RunOutputOptions controlling stdout/stderr handling.
timeout: Optional wall-clock timeout in seconds; None disables timeouts. Expiry terminates the subprocess exactly as run() does.
context: Optional execution settings such as env, cwd, and cancel grace.
stdin: Optional stdin source, as for run.

Returns

LineStream: An async iterator of LineEvent whose result attribute holds the final CommandResult once iteration completes.

Raises

ForbiddenProgramError: If the program is not permitted by the active context allowlist.
TimeoutExpired: If *timeout* elapses before the command completes.
UnicodeEncodeError: If stdin text cannot be encoded with the context's encoding.
StdinSourceError: If a StdinStream producer or its encoder fails.
ValueError: If *output* redirects stdout: line iteration reads stdout through a parent-side pipe, so a redirected stdout has nothing to iterate. Use run when stdout goes to a file or descriptor.

method SafeCmd.run_sync

def run_sync(
    *,
    output: RunOutputOptions | None = None,
    timeout: float | None = None,
    context: ExecutionContext | None = None,
    stdin: StdinSource | None = None,
) -> CommandResult

Execute the command synchronously.

Parameters

output RunOutputOptions | None, default=None: Capture and echo settings. The default limits each mirrored line to 64 KiB; max_echo_line_bytes=None restores unbounded echoing while preserving the same capture contract.
timeout float | None, default=None: Maximum execution time in seconds.
context ExecutionContext | None, default=None: Execution settings, including echo sinks and text encoding.
stdin StdinInput | StdinStream | None, default=None: Optional stdin source, as for run.

Returns

CommandResult: The command outcome, including complete captured streams when enabled.

Raises

PermissionError: If the command is not allowed by the active scope.
TimeoutError: If execution exceeds the effective timeout.
UnicodeEncodeError: If text stdin cannot be encoded by the execution context.
StdinSourceError: If a StdinStream producer or its encoder fails.

method SafeCmd.start

async def start(
    *,
    output: RunOutputOptions | None = None,
    timeout: float | None = None,
    context: ExecutionContext | None = None,
    stdin: StdinInput | None = None,
) -> RunningCommand

Start the command and return an owned handle to it.

The child is spawned and its PID announced before this returns; the caller decides what happens next. Await RunningCommand.wait for the outcome, deliver a signal with RunningCommand.signal, or stop the child with RunningCommand.terminate or RunningCommand.kill.

The handle owns the child's lifetime. Using it as an async with block settles the run on exit — a child still running is stopped down the same SIGTERM, cancel-grace, SIGKILL path a cancelled run() takes — so a caller cannot leave a child behind by forgetting about it. Output capture is always on, whatever *output* says, so a stopped child still reports what the command wrote.

Unlike run(), no deadline is enforced while the handle is merely held: the clock a timeout is measured against starts at the wait that asks for one. An explicit *timeout* here is therefore the default deadline for every wait() on this handle rather than a limit on how long the child may live.

Parameters

output RunOutputOptions | None, default=None: Capture and echo settings. Capture defaults to on, so the handle's result carries the command's streams.
timeout float | None, default=None: Default maximum time a wait() on this handle may take, in seconds. An explicit value overrides the timeout in context.
context ExecutionContext | None, default=None: Execution settings, including echo sinks, encoding, and the cancel grace a graceful stop allows.
stdin StdinInput | None, default=None: Optional bytes or text supplied to the child process's stdin.

Returns

RunningCommand: An owned handle to the running command.

Raises

PermissionError: If the command is not allowed by the active scope.
UnicodeEncodeError: If text stdin cannot be encoded by the execution context.

method SafeCmd.run_bytes

async def run_bytes(
    *,
    output: RunOutputOptions | None = None,
    timeout: float | None = None,
    context: ExecutionContext | None = None,
    stdin: StdinSource | None = None,
) -> BytesCommandResult

Execute the command, capturing its output as bytes.

The run itself is the one run() performs; only what happens to the captured streams differs. Bytes are returned exactly as the child wrote them — no decoding, no replacement characters — so values that are not valid text survive the round trip.

Echoing, sink presentation, and idle reporting all behave as they do in text mode. A sink exposing a buffer receives the child's own bytes; a text-only sink decodes them for display only, and a mirror it cannot render is disabled without disturbing what was captured.

Parameters

output RunOutputOptions | None, default=None: Capture and echo settings. on_line is rejected: bytes mode returns the child's own bytes, so a callback carrying decoded text would be a second, contradictory contract for the same stream. Use the text-mode entry point when lines are wanted.
timeout float | None, default=None: Maximum execution time in seconds. An explicit value overrides the timeout in context.
context ExecutionContext | None, default=None: Execution settings, including echo sinks and text encoding.
stdin StdinInput | StdinStream | None, default=None: Optional stdin source, as for run: a complete payload, a producer pulled one chunk at a time, or None to inherit the parent's stdin.

Returns

BytesCommandResult: The command outcome, carrying bytes for each captured stream when output.capture is true and None for each stream that was not captured.

Raises

ValueError: If output.on_line is set.
PermissionError: If the command is not allowed by the active scope.
TimeoutError: If execution exceeds the effective timeout.
UnicodeEncodeError: If text stdin cannot be encoded by the execution context.
StdinSourceError: If a StdinStream producer or its encoder fails.

method SafeCmd.run_bytes_sync

def run_bytes_sync(
    *,
    output: RunOutputOptions | None = None,
    timeout: float | None = None,
    context: ExecutionContext | None = None,
    stdin: StdinSource | None = None,
) -> BytesCommandResult

Execute the command synchronously, capturing its output as bytes.

Parameters

output RunOutputOptions | None, default=None: Capture and echo settings. on_line is rejected: bytes mode returns the child's own bytes, so a callback carrying decoded text would be a second, contradictory contract for the same stream. Use the text-mode entry point when lines are wanted.
timeout float | None, default=None: Maximum execution time in seconds.
context ExecutionContext | None, default=None: Execution settings, including echo sinks and text encoding.
stdin StdinInput | StdinStream | None, default=None: Optional stdin source, as for run: a complete payload, a producer pulled one chunk at a time, or None to inherit the parent's stdin.

Returns

BytesCommandResult: The command outcome, carrying bytes for each captured stream when capture is enabled and None for each stream that was not captured.

Raises

ValueError: If output.on_line is set.
PermissionError: If the command is not allowed by the active scope.
TimeoutError: If execution exceeds the effective timeout.
UnicodeEncodeError: If text stdin cannot be encoded by the execution context.
StdinSourceError: If a StdinStream producer or its encoder fails.

class

`SafeCmdBuilder`

cuprum/sh/builder.py

SafeCmdBuilder

Bases: typ.Protocol

Structural contract for the callable sh.make returns.

The protocol is deliberately not decorated with typing.runtime_checkable. It exists to describe a call signature to static checkers, and an isinstance test against a protocol whose only method is __call__ would confirm little beyond the object being callable at all.

type alias

`ArgValue`

cuprum/sh/argv.py

ArgValue = str | int | float | bool | Path | os.PathLike[str]

Values a sh.make builder accepts as positional or keyword arguments.

The alias is public so callers can annotate their own wrappers and fixtures with the same argument domain the builders enforce, without repeating the union or reaching for object. None is deliberately absent: a builder that receives it raises TypeError at call time.

os.PathLike[str] is named rather than the bare os.PathLike so the annotation states that only text-returning paths are accepted. The protocol is structural, so the runtime cannot enforce the str parameter from the annotation alone; _stringify_arg therefore re-checks the os.fspath result and rejects a __fspath__ that returns bytes.

class

`CommandResult`

cuprum/sh/results.py

class CommandResult(
    program: Program,
    argv: tuple[str, ...],
    exit_code: int,
    pid: int,
    stdout: str | None,
    stderr: str | None,
    started_at: float = dc.field(default=0.0, kw_only=True),
    duration: float = dc.field(default=0.0, kw_only=True),
    max_rss_bytes: int | None = dc.field(default=None, kw_only=True),
    user_cpu_seconds: float | None = dc.field(default=None, kw_only=True),
    system_cpu_seconds: float | None = dc.field(default=None, kw_only=True),
    resolved_path: str | None = dc.field(default=None, kw_only=True),
    relay_fallbacks: tuple[RelayFallback, ...] = (),
)

Structured result returned by command execution.

Attributes

program: Program that was executed, as the catalogue identity policy was checked against.
resolved_path: The executable that actually ran, when a scope bound program to a specific path. None when no binding applied, in which case the child ran under the catalogued name. Filesystem replacement after validation is not detected; see cuprum.executable_binding.
argv: Argument vector (excluding the program name) passed to the process.
exit_code: Exit status reported by the process.
pid: Process identifier; -1 when unavailable.
stdout: Captured standard output, or None when capture was disabled.
stderr: Captured standard error, or None when capture was disabled.
started_at: Wall-clock timestamp at which process execution started.
duration: Monotonic process duration in seconds.
max_rss_bytes: Peak resident set size of the executed child in bytes. Direct commands on Linux and macOS report this from the platform wait4 call, which attributes the figure to that one child; Linux reports KiB and macOS bytes, both normalized to bytes here. It is never derived from the process-global RUSAGE_CHILDREN high-water mark, which cannot be attributed safely to one command. None on Windows, on platforms without the child-specific interface, and for every pipeline stage, whose concurrently reaped children cannot be separated.
user_cpu_seconds: User CPU time consumed by the executed child in seconds. Direct commands on Linux and macOS report this from wait4; elsewhere the aggregate RUSAGE_CHILDREN fallback may supply it, and those deltas are approximate under run_concurrent. None on Windows, on platforms without child resource accounting, and for pipeline stages.
system_cpu_seconds: System CPU time consumed by the executed child in seconds. Direct commands on Linux and macOS report this from wait4; elsewhere the aggregate RUSAGE_CHILDREN fallback may supply it, and those deltas are approximate under run_concurrent. None on Windows, on platforms without child resource accounting, and for pipeline stages.
relay_fallbacks: Handled echo-disablement records from this command's own streams, one per affected drain in stdout-then-stderr order. The ordering does not reconstruct chronological interleaving between the two streams. Empty when no echo write was handled as failed, and an echo failure does not change exit_code or ok.

14 members

field CommandResult.program

program: Program

field CommandResult.argv

argv: tuple[str, ...]

field CommandResult.exit_code

exit_code: int

field CommandResult.pid

pid: int

field CommandResult.stdout

stdout: str | None

field CommandResult.stderr

stderr: str | None

field CommandResult.started_at

started_at: float

field CommandResult.duration

duration: float

field CommandResult.max_rss_bytes

max_rss_bytes: int | None

field CommandResult.user_cpu_seconds

user_cpu_seconds: float | None

field CommandResult.system_cpu_seconds

system_cpu_seconds: float | None

field CommandResult.resolved_path

resolved_path: str | None

field CommandResult.relay_fallbacks

relay_fallbacks: tuple[RelayFallback, ...]

property CommandResult.ok

ok: bool

Whether the command exited successfully.

Returns

bool: True exactly when exit_code is zero.

class

`BytesCommandResult`

cuprum/sh/results.py

class BytesCommandResult(
    program: Program,
    argv: tuple[str, ...],
    exit_code: int,
    pid: int,
    stdout: bytes | None,
    stderr: bytes | None,
    started_at: float = dc.field(default=0.0, kw_only=True),
    duration: float = dc.field(default=0.0, kw_only=True),
    max_rss_bytes: int | None = dc.field(default=None, kw_only=True),
    user_cpu_seconds: float | None = dc.field(default=None, kw_only=True),
    system_cpu_seconds: float | None = dc.field(default=None, kw_only=True),
    resolved_path: str | None = dc.field(default=None, kw_only=True),
    relay_fallbacks: tuple[RelayFallback, ...] = (),
)

Byte-exact result returned by the binary-output command entry points.

The fields, defaults, and helpers mirror CommandResult; only the captured-output fields differ, so a caller that switches entry point switches the static type of what it reads rather than the shape of the record. Bytes reach a caller untouched — no decoding, no surrogate escape — which is what lets every byte value survive the round trip.

Attributes

program: Program that was executed.
argv: Argument vector (excluding the program name) passed to the process.
exit_code: Exit status reported by the process.
pid: Process identifier; -1 when unavailable.
stdout: Captured standard output bytes, or None when capture was disabled. A stream that was consumed but produced nothing is b"", which is distinguishable from the None of a run that never captured.
stderr: Captured standard error bytes, with the same None/b"" distinction as stdout.
started_at: Wall-clock timestamp at which process execution started.
duration: Monotonic process duration in seconds.
max_rss_bytes: Peak resident set size of the executed child in bytes, under the same platform rules as CommandResult.
user_cpu_seconds: User CPU time consumed by the executed child in seconds, under the same platform rules as CommandResult.
system_cpu_seconds: System CPU time consumed by the executed child in seconds, under the same platform rules as CommandResult.
resolved_path: The executable that actually ran, under the same rules as CommandResult.
relay_fallbacks: Handled echo-disablement records from this command's own streams, in the order and with the meaning CommandResult documents.

14 members

field BytesCommandResult.program

program: Program

field BytesCommandResult.argv

argv: tuple[str, ...]

field BytesCommandResult.exit_code

exit_code: int

field BytesCommandResult.pid

pid: int

field BytesCommandResult.stdout

stdout: bytes | None

field BytesCommandResult.stderr

stderr: bytes | None

field BytesCommandResult.started_at

started_at: float

field BytesCommandResult.duration

duration: float

field BytesCommandResult.max_rss_bytes

max_rss_bytes: int | None

field BytesCommandResult.user_cpu_seconds

user_cpu_seconds: float | None

field BytesCommandResult.system_cpu_seconds

system_cpu_seconds: float | None

field BytesCommandResult.resolved_path

resolved_path: str | None

field BytesCommandResult.relay_fallbacks

relay_fallbacks: tuple[RelayFallback, ...]

property BytesCommandResult.ok

ok: bool

Whether the command exited successfully.

Returns

bool: True exactly when exit_code is zero.

class

`RunningCommand`

cuprum/_process_handle/state.py

class RunningCommand(run: _HandleRun) -> None

An owned handle to a command that is already running.

Returned by SafeCmd.start. Unlike SafeCmd.run, which awaits a child to completion, this handle keeps the child alive and hands the caller explicit control over it: wait for the outcome, signal to deliver a POSIX signal, terminate for the graceful stop, and kill for the immediate one.

The handle owns the child's lifetime. Leaving an async with block settles it deterministically — a child still running is stopped down the same SIGTERM, cancel-grace, SIGKILL path a cancelled run() takes — so no child is left running or unreaped because a caller forgot to stop it.

Every operation that settles is idempotent and safe to repeat, and none of them suppresses the exception of an enclosing block. Repeating is safe on both halves of the operation: the outcome the settlement records is shared, and the stop that produces it is serialized, so two callers stopping the same child escalate it once between them however they interleave. Once the child has exited, result holds the same CommandResult a run() call would have returned, and further waits or terminations report that outcome rather than failing.

A deadline is the one exception, and it is raised once. The wait whose clock ran out reports TimeoutExpired after stopping and reaping the child; because that work is done, the outcome it produced is recorded, and a later wait, termination, or teardown returns it instead of re-raising an error about a child that is already gone.

7 members

property RunningCommand.pid

pid: int | None

The child's process identifier, as announced by its start event.

Returns

int | None: The child's PID, or None on a platform whose transport never reported one.

property RunningCommand.exec_id

exec_id: ExecId

The execution identifier every event from this run carries.

Returns

ExecId: The identifier shared by the run's plan, start, exit, and settled events, so a consumer can correlate the handle with the observation stream that describes it.

property RunningCommand.result

result: CommandResult | None

The settled outcome, or None while the command is still running.

Returns

CommandResult | None: The command's result once settlement has completed, otherwise None. The property never waits; use wait to block until there is an outcome to read. A wait whose deadline elapsed publishes the outcome here too, because the child it stopped is nonetheless reaped and has one.

method RunningCommand.wait

async def wait(*, timeout: float | None = None) -> CommandResult

Wait for the command to finish and return its outcome.

The first caller to arrive settles the run; every later caller awaits the same settlement and sees the same outcome, so calling this repeatedly or concurrently is safe and reports one result.

Parameters

timeout float | None, default=None: Seconds to allow the command to finish. Expiry stops the child the way a timed-out run() does and raises TimeoutExpired carrying the output captured before the deadline. None falls back to the deadline SafeCmd.start resolved for this handle, and to no deadline at all when start() resolved none.

Returns

CommandResult: The command's outcome once it has exited.

Raises

TimeoutExpired: If *timeout*, or the deadline start() resolved, elapsed before the command finished.

method RunningCommand.signal

async def signal(signal_number: int) -> None

Deliver a POSIX signal to the running command.

Unlike terminate and kill, this asks for exactly the signal named and takes no view on how the child should respond. It does not settle the run: a child sent SIGSTOP, or sent a signal it ignores, is still the caller's to wait on or stop.

Parameters

signal_number int: The signal to deliver, as signal spells it.

Raises

SignalAfterExitError: If the command has already exited, carrying its exit code.
UnsupportedSignalError: If the platform cannot deliver a caller-named signal.

method RunningCommand.terminate

async def terminate(*, grace: float | None = None) -> CommandResult

Stop the command gracefully, escalating if it does not comply.

The child is sent SIGTERM, given *grace* seconds, then sent SIGKILL if it is still there. Both the signal that ended it and the output it produced before ending are preserved: a child that exited on the SIGTERM reports -15, and one that had to be killed reports -9.

Calling this on a command that has already exited is a no-op that reports the recorded outcome — including when a wait deadline is what ended the child, since the outcome that wait produced is the one this returns.

Parameters

grace float | None, default=None: Seconds to wait between the graceful signal and the forced one. Defaults to the execution context's cancel grace, which is the same window a cancelled run() allows.

Returns

CommandResult: The ended command's outcome.

method RunningCommand.kill

async def kill() -> CommandResult

Stop the command immediately, without a grace period.

This is the forced half of terminate on its own: the child is sent SIGKILL and reaped, so its outcome reports -9 unless it had already exited, in which case the recorded outcome is returned. Unlike terminate, no SIGTERM is sent first, so the reported signal is the one the caller asked for rather than whichever one happened to end the child.

Returns

CommandResult: The ended command's outcome.

class

`ExecutionContext`

cuprum/sh/execution.py

class ExecutionContext(
    env: _EnvMapping = None,
    cwd: _CwdType = None,
    cancel_grace: float = _DEFAULT_CANCEL_GRACE,
    native_pump_cleanup_grace: float = _DEFAULT_NATIVE_PUMP_CLEANUP_GRACE,
    timeout: float | None = None,
    stdout_sink: typ.IO[str] | None = None,
    stderr_sink: typ.IO[str] | None = None,
    encoding: str = _DEFAULT_ENCODING,
    errors: str = _DEFAULT_ERROR_HANDLING,
    tags: cabc.Mapping[str, object] | None = None,
    env_mode: EnvMode = EnvMode.OVERLAY,
    process_group: ProcessGroupPolicy | str = ProcessGroupPolicy.INHERIT,
)

Execution parameters for SafeCmd runtime control.

Attributes

env: Environment variable overlay applied to the subprocess.
cwd: Working directory for the subprocess.
cancel_grace: Seconds to wait after SIGTERM before escalating to SIGKILL.
native_pump_cleanup_grace: Seconds to wait for a cancelled native-pump worker before its descriptor cleanup is deferred to its completion callback.
timeout: Optional runtime timeout in seconds. None means no override.
stdout_sink: Text sink for echoing stdout; defaults to the active sys.stdout.
stderr_sink: Text sink for echoing stderr; defaults to the active sys.stderr. When no on_idle callback is supplied, it also receives the idle heartbeat's keepalive line, written and flushed synchronously on the run's event loop, so its write and flush must return promptly: a sink that blocks delays the run's stream reads, timeout handling, and cancellation. Hand a slow destination to a worker thread, an executor, or a genuinely non-blocking drain such as a queue fed with put_nowait. A separate asyncio task on the run's own loop is not enough: draining that queue still competes with the parent's stream reads.
encoding: Character encoding used when decoding subprocess output.
errors: Error handling strategy applied during decoding.
tags: Optional metadata attached to structured execution events.
env_mode: Environment policy applied when rendering env for the subprocess.
process_group: Whether the run owns the child's process group. INHERIT keeps the child in the parent's group and signals only the direct child on teardown; OWN_GROUP spawns the child as the leader of its own session and process group on POSIX and terminates that whole group. A member may be given either as a ProcessGroupPolicy or as the string spelling of one, which __post_init__ resolves to the member it names and rejects if it names none. See ProcessGroupPolicy.

13 members

field ExecutionContext.env

env: _EnvMapping

field ExecutionContext.cwd

cwd: _CwdType

field ExecutionContext.cancel_grace

cancel_grace: float

field ExecutionContext.native_pump_cleanup_grace

native_pump_cleanup_grace: float

field ExecutionContext.timeout

timeout: float | None

field ExecutionContext.stdout_sink

stdout_sink: typ.IO[str] | None

field ExecutionContext.stderr_sink

stderr_sink: typ.IO[str] | None

field ExecutionContext.encoding

encoding: str

field ExecutionContext.errors

errors: str

field ExecutionContext.tags

tags: cabc.Mapping[str, object] | None

field ExecutionContext.env_mode

env_mode: EnvMode

field ExecutionContext.process_group

process_group: ProcessGroupPolicy | str

property ExecutionContext.resolved_process_group

resolved_process_group: ProcessGroupPolicy

The policy behind the declared ProcessGroupPolicy | str field.

__post_init__ normalizes the field, but the declared type stays wide because a caller may spell the policy as a string. This is the narrow view the execution layer reads, so it never has to parse.

Returns

ProcessGroupPolicy: The normalized policy member, never a string.

class

`ProcessGroupPolicy`

cuprum/sh/execution.py

ProcessGroupPolicy

Bases: enum.StrEnum

Choose whether a run owns the process group its child starts.

INHERIT is the default. The child joins the parent's process group and session, and teardown signals the direct child alone, so a run terminates exactly the process it spawned.

OWN_GROUP gives the run ownership of a process group of its own. On POSIX the child is spawned with start_new_session=True, which makes it the leader of a new session and a new process group, so the group is addressed by the child's own process identifier. Teardown then signals every member of that group, which lets a child's descendants reach end-of-file on inherited pipes instead of holding a timed-out run's drain open.

Ownership is bounded. A descendant that leaves the group deliberately — with setsid() or setpgid() — is outside it and outside the guarantee, as are the run's ancestors, its siblings, and any process that predates the spawn. Callers needing containment for a descendant that detaches its own session require a supervisor, a control group, or a container.

OWN_GROUP is rejected with a ValueError at spawn on Windows, which has no POSIX process group and no way to assign a Job Object atomically with process creation.

2 members

member ProcessGroupPolicy.INHERIT

INHERIT = 'inherit'

member ProcessGroupPolicy.OWN_GROUP

OWN_GROUP = 'own_group'

class

`RunOutputOptions`

cuprum/sh/output.py

class RunOutputOptions(
    capture: bool = True,
    echo: bool = False,
    echo_stdout: bool | None = None,
    echo_stderr: bool | None = None,
    max_echo_line_bytes: int | None = DEFAULT_ECHO_MAX_LINE_BYTES,
    broken_pipe_policy: BrokenPipePolicy | str = dc.field(default=BrokenPipePolicy.STRICT, kw_only=True),
    on_line: LineHook | None = None,
    idle_after: float | None = None,
    on_idle: cabc.Callable[[float, float], None] | None = None,
    sink: sinks.OutputSink | None = None,
    group: bool = False,
    annotate_failure: bool = False,
    stdin: StdioTarget | None = None,
    stdout: StdioTarget | None = None,
    stderr: StdioTarget | None = None,
    _synthesized_sink: sinks.OutputSink | None = dc.field(default=None, repr=False, compare=False, kw_only=True),
)

Configure captured and mirrored command output.

Parameters

capture bool, default=True: Store stdout and stderr on the returned result. Capture is independent of echoing, so a captured stream can remain silent and an echoed stream can be left uncaptured.
echo bool, default=False: Shorthand for enabling both echo_stdout and echo_stderr unless either stream has an explicit override.
echo_stdout bool | None, default=None: Whether stdout is mirrored to the execution context's stdout sink. None inherits echo.
echo_stderr bool | None, default=None: Whether stderr is mirrored to the execution context's stderr sink. None inherits echo.
max_echo_line_bytes int | None, default=64 * 1024: Inclusive byte bound for every echoed line, including its retained bytes, truncation marker, and terminator. None restores unbounded, chunk-for-chunk mirroring; captured output always remains complete.
broken_pipe_policy BrokenPipePolicy | str, keyword-only, default=STRICT: How echoing responds when a presentation sink reports BrokenPipeError, which is what a destination that has closed under the run looks like from inside the drain. STRICT propagates the error and aborts the run, preserving the behaviour of every caller that does not name a policy. BEST_EFFORT disables echoing for the affected stream only, so capture, line observation, and child reaping continue and the caller still receives a result; the transition is reported once through CommandResult.relay_fallbacks, the echo observation channel, and a structured cuprum.stream warning. Either the member or its string value is accepted, and anything else raises ValueError. Only BrokenPipeError is affected: any other sink OSError propagates under both policies, so an unreachable device is never mistaken for a closed reader. The field is keyword-only because it was inserted mid-list: accepting it positionally would have taken on_line's slot and shifted every field after it.
on_line LineHook | None, default=None: Optional synchronous callback invoked once per decoded output line with a LineEvent carrying the stream name, the monotonic seconds since the command started, and the line text. Independent of capture and echo; lines are delivered in arrival order per stream. Lines are observed on the Python pathway, so the Rust fast-path dispatcher stays out of the way whenever this is set.
idle_after float | None, default=None: Seconds of silence, measured across both streams, before the run reports that it is still running. None disables idle reporting and costs nothing: no watchdog, no timer, no extra pipe. The interval must be finite and strictly positive. Reporting describes the absence of observed output — never a deadlock diagnosis — and can neither terminate the child nor extend its timeout.
on_idle cabc.Callable[[float, float], None] | None, default=None: Synchronous (elapsed_total, elapsed_idle) callback, in seconds, invoked once per idle interval in place of the built-in stderr keepalive. It must not block for long: it runs on the run's own event loop. Requires idle_after.
stdin StdioTarget | None, default=None: Where the child's standard input is bound, for the two variants that describe the stream itself. StdioTarget.pipe() asks for a library-owned pipe that nothing writes to, so a child reading stdin sees EOF instead of the parent's terminal; StdioTarget.inherit() asks for the parent's stdin, which is the default. The actual input is supplied through SafeCmd.run's stdin= argument --- a StdinInput payload or a StdinStream producer --- and either form implies a pipe. A path or fd target is rejected here: an input descriptor belongs to the source, not to these options.
sink sinks.OutputSink | None, default=None: Optional presentation adapter (sinks protocol). When given, it may reframe the parent-facing output of this run (for example, a GitHub Actions group); the default None keeps the plain two-stream behaviour. An adapter is inactive for a run when it declines activation, in which case output is unchanged.
group bool, default=False: Convenience opt-in for a GitHub Actions collapsible log group around this run's echoed output. It is a shim over sink: setting it constructs GitHubActionsSink(emit_group=True) and stores it in sink when the caller supplied none, so the framing, the stop-commands lease that shields it from child output, and the workflow-command syntax all stay inside the adapter and the execution layer learns nothing new. A pipeline emits one group for the whole pipeline, not one per stage. Values other than bool raise ValueError.
annotate_failure bool, default=False: Convenience opt-in for a single ::error:: annotation when the run ends in a non-zero exit, a timeout, or an error. Synthesizes GitHubActionsSink(emit_annotation=True) the same way group does, and composes with it: either flag alone synthesizes the adapter with only its own half enabled. Values other than bool raise ValueError.

Notes

The two convenience flags reuse the same environment gate as any other sink: the adapter they construct carries no force, so they take effect only when the parent process runs on GitHub Actions (GITHUB_ACTIONS == "true"). Outside that environment they synthesize a sink that declines activation and the run keeps its plain destinations byte-for-byte — which is the intended behaviour, not a failure to apply them. Callers who want the framing locally, on a non-standard runner, or through a custom destination or title pass sink= explicitly, possibly with force=True.

An explicit sink always wins: when one is supplied the flags are recorded but ignored, and no adapter is synthesized. Put the sink on a new RunOutputOptions or use dataclasses.replace when adapting shared options; run methods do not override output.sink per call. Passing a sink obtained from another options object to a new RunOutputOptions(sink=...) also makes it explicit, even when the sink was originally synthesized by the convenience flags. Replacing either flag on flag-generated options rebuilds their adapter to match the new values, while an explicitly supplied sink remains untouched.

The default GitHub Actions sink does not serialize overlapping sessions. Do not run grouped commands concurrently when they write to the same parent stderr; run them sequentially so their workflow frames cannot interleave.

Workflow commands are written to the parent's stderr by default. Neither flag changes capture, exit codes, or the returned result.

Examples

>>> options = RunOutputOptions(capture=True, echo=True)
>>> options.resolved_echo
(True, True)
>>> RunOutputOptions(capture=True, echo=True, echo_stdout=False).resolved_echo
(False, True)
>>> RunOutputOptions(capture=False, idle_after=30.0).capture
False
>>> RunOutputOptions(group=True).sink is not None
True
>>> RunOutputOptions().sink is None
True

17 members

field RunOutputOptions.capture

capture: bool

field RunOutputOptions.echo

echo: bool

field RunOutputOptions.echo_stdout

echo_stdout: bool | None

field RunOutputOptions.echo_stderr

echo_stderr: bool | None

field RunOutputOptions.max_echo_line_bytes

max_echo_line_bytes: int | None

field RunOutputOptions.broken_pipe_policy

broken_pipe_policy: BrokenPipePolicy | str

field RunOutputOptions.on_line

on_line: LineHook | None

field RunOutputOptions.idle_after

idle_after: float | None

field RunOutputOptions.on_idle

on_idle: cabc.Callable[[float, float], None] | None

field RunOutputOptions.sink

sink: sinks.OutputSink | None

field RunOutputOptions.group

group: bool

field RunOutputOptions.annotate_failure

annotate_failure: bool

field RunOutputOptions.stdin

stdin: StdioTarget | None

field RunOutputOptions.stdout

stdout: StdioTarget | None

field RunOutputOptions.stderr

stderr: StdioTarget | None

property RunOutputOptions.resolved_broken_pipe_policy

resolved_broken_pipe_policy: BrokenPipePolicy

The policy behind the declared BrokenPipePolicy | str field.

__post_init__ normalizes the field, but the declared type stays wide because a caller may spell the policy as a string. This is the narrow view the execution layer reads, so it never has to parse.

Returns

BrokenPipePolicy: The normalized policy member, never a string and never None.

property RunOutputOptions.resolved_echo

resolved_echo: tuple[bool, bool]

The resolved (echo_stdout, echo_stderr) gates.

__post_init__ fills any None per-stream field from echo, so the returned pair is always concrete and reflects an explicit per-stream override where one was supplied.

class

`BrokenPipePolicy`

cuprum/echo_events.py

BrokenPipePolicy

Bases: enum.StrEnum

How a drain responds when an echo sink reports a broken pipe.

A presentation sink writes to a destination the parent does not own — a log collector, a console, another process's reader — and that destination can close before the child stops writing. The BrokenPipeError that follows is indistinguishable, from inside the drain, from any other sink failure, so the caller names the intended response rather than the drain guessing.

Members ------- STRICT Propagate BrokenPipeError out of the drain. This is the default, and preserves the behaviour every existing caller already has. BEST_EFFORT Disable echoing for the affected stream only, so capture, line observation, and child reaping continue. The transition is reported once through the echo observation channel, the cuprum.stream WARNING, and the command's result diagnostics.

Only BrokenPipeError is affected. A sink that fails with any other OSError still propagates under both members, so a genuinely unreachable device is never mistaken for a closed reader.

Examples

The member value is the string operators see::

    assert BrokenPipePolicy.BEST_EFFORT == "best_effort"

2 members

member BrokenPipePolicy.STRICT

STRICT = 'strict'

member BrokenPipePolicy.BEST_EFFORT

BEST_EFFORT = 'best_effort'

class

`IOOptions`

cuprum/sh/output.py

class IOOptions()

Bases: RunOutputOptions

Deprecated alias for command output stream options.

17 members

field IOOptions.capture

capture: bool

field IOOptions.echo

echo: bool

field IOOptions.echo_stdout

echo_stdout: bool | None

field IOOptions.echo_stderr

echo_stderr: bool | None

field IOOptions.max_echo_line_bytes

max_echo_line_bytes: int | None

field IOOptions.broken_pipe_policy

broken_pipe_policy: BrokenPipePolicy | str

field IOOptions.on_line

on_line: LineHook | None

field IOOptions.idle_after

idle_after: float | None

field IOOptions.on_idle

on_idle: cabc.Callable[[float, float], None] | None

field IOOptions.sink

sink: sinks.OutputSink | None

field IOOptions.group

group: bool

field IOOptions.annotate_failure

annotate_failure: bool

field IOOptions.stdin

stdin: StdioTarget | None

field IOOptions.stdout

stdout: StdioTarget | None

field IOOptions.stderr

stderr: StdioTarget | None

property IOOptions.resolved_broken_pipe_policy

resolved_broken_pipe_policy: BrokenPipePolicy

The policy behind the declared BrokenPipePolicy | str field.

__post_init__ normalizes the field, but the declared type stays wide because a caller may spell the policy as a string. This is the narrow view the execution layer reads, so it never has to parse.

Returns

BrokenPipePolicy: The normalized policy member, never a string and never None.

property IOOptions.resolved_echo

resolved_echo: tuple[bool, bool]

The resolved (echo_stdout, echo_stderr) gates.

__post_init__ fills any None per-stream field from echo, so the returned pair is always concrete and reflects an explicit per-stream override where one was supplied.

class

`StdinInput`

cuprum/sh/execution.py

class StdinInput(text: str | None = None, data: bytes | None = None)

Caller-provided data to write to a subprocess's stdin pipe.

Exactly one of *text* or *data* may be supplied.

3 members

field StdinInput.text

text: str | None

field StdinInput.data

data: bytes | None

method StdinInput.resolve

def resolve(ctx: ExecutionContext) -> bytes | None

Return the bytes payload, encoding *text* with *ctx* when needed.

Parameters

ctx ExecutionContext: The execution context whose encoding and errors encode text when no raw data is set.

Returns

bytes | None: The raw *data* payload, or *text* encoded with ctx.encoding and ctx.errors; None when neither field is set.

Raises

UnicodeEncodeError: If text cannot be encoded with ctx.encoding under ctx.errors (for example, errors="strict").

class

`StdinStream`

cuprum/sh/execution.py

class StdinStream(
    chunks: cabc.AsyncIterable[str | bytes] | cabc.AsyncIterator[str | bytes],
)

A library-owned, pull-after-drain producer for stdin.

Cuprum pulls one chunk from *chunks*, writes it to the child's stdin pipe, and waits for that write to drain before pulling the next. That is what lets a caller feed a child more data than they would ever hold in one buffer: the payload is never materialized whole. What it bounds is how far *ahead* the producer runs, not the size of a single chunk. While drain() waits, the chunk just pulled and its encoded payload are still in memory alongside the transport's write buffer and the OS pipe, so retained memory is roughly the largest chunk the producer yields plus those buffers. A caller who cares about peak memory should yield bounded-size chunks.

The producer is advanced with aiter(), so *chunks* may be any async iterable. str chunks are encoded with the run's ExecutionContext.encoding and ExecutionContext.errors, incrementally, so a multi-byte character split across two chunks is still encoded correctly. bytes chunks are written verbatim.

Cuprum owns the producer for the duration of the run. On every exit path -- normal completion, producer failure, exceeding the timeout, and cancellation -- it closes the child's pipe and then calls the iterator's aclose() when the producer provides one. That order is deliberate: the pipe is cuprum's own resource and the child's EOF depends on the close reaching it, so releasing it first means a caller's cleanup, however slow, cannot hold the child's input open.

Both steps are bounded. Closing the pipe gives a graceful shutdown a short window and then releases the transport unconditionally, because a writer whose grandchild holds the read end open would otherwise wait forever. The producer's aclose() gets the same window: a finalizer that suspends longer than that is *cancelled*, not waited out, and one that absorbs cancellation keeps working in the background. So cuprum bounds its own wait rather than promising that a caller's cleanup has completed. A caller therefore does not need to finalize their own generator; a caller must not yield from a generator that something else is concurrently driving.

A child that closes its stdin early (head is the canonical example) is normal rather than an error: the resulting BrokenPipeError is recorded as a stdin_error observation and the run continues to its exit code. Only producer and encoder failures raise StdinSourceError.

1 member

field StdinStream.chunks

chunks: cabc.AsyncIterable[str | bytes] | cabc.AsyncIterator[str | bytes]

The async iterable or iterator whose chunks become the child's stdin.

type alias

`StdinSource`

cuprum/sh/execution.py

StdinSource = StdinInput | StdinStream

Either a complete payload or a streaming producer.

class

`StdioTarget`

cuprum/sh/stdio.py

class StdioTarget(
    kind: _StdioKind = 'pipe',
    value: Path | int | typ.IO[bytes] | typ.IO[str] | None = None,
)

Where a child's standard stream is bound.

The four variants differ in *ownership*, which is the whole contract:

- pipe — a library-owned pipe, the default for stdout and stderr whenever capture, echo, idle reporting, or line observation needs it. Cuprum creates it, consumes or writes it, and closes it. - inherit — the parent's stream, passed straight through. This is the default for stdin, and the explicit way to let an unobserved stdout or stderr through: an unset stdout or stderr whose pipe nobody reads resolves to /dev/null instead, so the child cannot write into the caller's terminal. Cuprum touches nothing either way. - path — a file **cuprum owns**. The file is opened immediately before the spawn, its descriptor is handed to the child, and cuprum's copy is closed in a finally right after the spawn returns, so the descriptor never outlives the run. The caller names a path and never manages a descriptor. Parent and child write at independent offsets because the descriptor starts at offset 0, so redirecting two streams to one path interleaves them and is rejected. - fd — a **borrowed** descriptor or open file object. The caller owns it: cuprum never closes it, and a borrowed file object is flushed before spawn so buffered caller-side bytes reach the child. A caller who reuses one descriptor across two runs shares a single file offset, so the second run writes wherever the first left off.

Instances are values: two targets built the same way compare equal, so RunOutputOptions stays value-like and dataclasses.replace behaves.

9 members

field StdioTarget.kind

kind: _StdioKind

field StdioTarget.value

value: Path | int | typ.IO[bytes] | typ.IO[str] | None

property StdioTarget.path_value

path_value: Path

The path this target owns.

Returns

Path: The file cuprum opens before spawn and closes immediately after.

Raises

ValueError: If this target is not a path variant.

property StdioTarget.fd_value

fd_value: int | typ.IO[bytes] | typ.IO[str]

The borrowed descriptor or file object this target names.

Returns

int | IO[bytes] | IO[str]: The caller-owned resource cuprum will never close.

Raises

ValueError: If this target is not an fd variant.

property StdioTarget.is_owned_path

is_owned_path: bool

Whether this target names a file cuprum must open and close.

staticmethod StdioTarget.pipe

def pipe() -> StdioTarget

Return the library-owned pipe variant.

Returns

StdioTarget: A target selecting a pipe cuprum creates and owns.

staticmethod StdioTarget.inherit

def inherit() -> StdioTarget

Return the inherit-the-parent's-stream variant.

Returns

StdioTarget: A target selecting the parent's own stream.

staticmethod StdioTarget.path

def path(path: str | Path) -> StdioTarget

Return the cuprum-owned file variant.

Parameters

path str | Path: The file to open before spawn and close immediately afterwards. Cuprum does not create intermediate directories, so the parent must already exist.

Returns

StdioTarget: A target naming the file cuprum will own for the run.

staticmethod StdioTarget.fd

def fd(fd: int | typ.IO[bytes] | typ.IO[str]) -> StdioTarget

Return the borrowed descriptor or file-object variant.

Parameters

fd int | IO[bytes] | IO[str]: A descriptor or an open file object the *caller* owns. Cuprum never closes it; a file object is flushed immediately before the spawn so buffered caller-side bytes reach the child.

Returns

StdioTarget: A target borrowing the caller's descriptor or file object.

class

`LineStream`

cuprum/_line_iteration.py

class LineStream(
    iterator: cabc.AsyncGenerator[LineEvent | CommandResult, None],
) -> None

Async iterator of LineEvent and the run's final CommandResult.

Returned by SafeCmd.lines. Iteration yields every decoded output line as it arrives; after the run completes, result holds the same CommandResult a run() call would have returned, including captured output when capture=True.

Leaving the loop early does not, on its own, stop the subprocess. async for never closes a custom iterator on break, and the driver keeps draining the child for as long as the stream is reachable. Use the stream as an async context manager, or call aclose explicitly, to guarantee teardown; dropping the last reference also closes it, but only when the event loop finalizes the generator, so that is not a guarantee.

Either teardown path ends the child the way a cancelled run() does: SIGTERM, the cancel grace wait, then SIGKILL.

1 member

method LineStream.aclose

async def aclose() -> None

Close the underlying generator, tearing the run down.

class

`LineEvent`

cuprum/lines.py

class LineEvent(stream: LineStreamName, at: float, text: str)

One decoded output line, tagged with its stream and arrival time.

Attributes

stream: Which stream the line arrived on: "stdout" or "stderr".
at: Monotonic seconds since the command started, read from the time.perf_counter clock at the moment the line was decoded. Values are only comparable within one command's run.
text: Decoded line content without its line terminator.

3 members

field LineEvent.stream

stream: LineStreamName

field LineEvent.at

at: float

field LineEvent.text

text: str

type alias

`LineHook`

cuprum/lines.py

LineHook = cabc.Callable[[LineEvent], None]

type alias

`LineStreamName`

cuprum/lines.py

LineStreamName = typ.Literal['stdout', 'stderr']

module

`builders`

cuprum/builders/__init__.py

from cuprum import builders

Core builders for common command-line tools.

13 exports

class builders.Compression

Compression

Compression algorithm for tar_create.

Modelling compression as a single enum makes the "exactly one algorithm" invariant unrepresentable as an invalid state: the previous design used three mutually-exclusive booleans that had to be reconciled at runtime.

Members ------- NONE No compression (the default); tar_create emits no flag. GZIP gzip compression (-z). BZIP2 bzip2 compression (-j). XZ xz compression (-J).

new type builders.GitRef

GitRef = typ.NewType('GitRef', str)

class builders.RsyncOptions

class RsyncOptions(
    archive: bool = False,
    delete: bool = False,
    dry_run: bool = False,
    verbose: bool = False,
    compress: bool = False,
    allow_relative: bool = False,
)

Optional flags for the rsync builder.

new type builders.SafePath

SafePath = typ.NewType('SafePath', str)

class builders.TarCreateOptions

class TarCreateOptions(
    compression: Compression = Compression.NONE,
    allow_relative: bool = False,
)

Optional flags for tar_create.

Attributes

compression Compression: Compression algorithm for the archive. Defaults to Compression.NONE.
allow_relative bool: Allow relative archive and source paths. This applies to both archive and source paths. Defaults to False.

function builders.git_checkout

def git_checkout(
    ref: str,
    *,
    create_branch: bool = False,
    detach: bool = False,
    force: bool = False,
) -> SafeCmd

Build a git checkout command with validated refs.

Parameters

ref str: Reference to check out.
create_branch bool: When True, create a new branch at the reference.
detach bool: When True, check out in detached HEAD state.
force bool: When True, force the checkout, overwriting local changes.

Returns

SafeCmd: The assembled git checkout command.

Raises

TypeError: If ref is not a str (propagated from git_ref).
ValueError: If both create_branch and detach are True, or if ref is not a valid git reference (the latter propagated from git_ref).

function builders.git_ref

def git_ref(value: str) -> GitRef

Validate a git ref name or object name.

Parameters

value: Ref value to validate.

Returns

GitRef: Validated ref string.

Raises

TypeError: If value is not a str.
ValueError: If value is not a valid git reference (propagated from _validate_git_ref).

function builders.git_rev_parse

def git_rev_parse(ref: str) -> SafeCmd

Build a git rev-parse command with validated refs.

Parameters

ref str: Reference to resolve.

Returns

SafeCmd: The assembled git rev-parse command.

Raises

TypeError: If ref is not a string.
ValueError: If ref is not a valid git reference.

function builders.git_status

def git_status(*, short: bool = False, branch: bool = False) -> SafeCmd

Build a git status command.

Parameters

short bool: When True, request short-format output.
branch bool: When True, include branch information.

Returns

SafeCmd: The assembled git status command.

function builders.rsync_sync

def rsync_sync(
    source: str | Path,
    destination: str | Path,
    *,
    options: RsyncOptions | None = None,
) -> SafeCmd

Build an rsync synchronization command.

Parameters

source: Source path to synchronize from.
destination: Destination path to synchronize to.
options: Optional rsync settings. When omitted, the command uses RsyncOptions().

Returns

SafeCmd: Safe command wrapper for the curated rsync program and generated argv.

function builders.safe_path

def safe_path(
    value: str | Path,
    *,
    allow_relative: bool = False,
) -> SafePath

Validate and normalize a filesystem path.

Parameters

value: Path value to validate.
allow_relative: When True, relative paths are permitted. Defaults to False.

Returns

SafePath: Normalized path string.

function builders.tar_create

def tar_create(
    archive: str | Path,
    sources: cabc.Sequence[str | Path],
    *,
    options: TarCreateOptions | None = None,
) -> SafeCmd

Build a tar create command.

Parameters

archive: Archive path passed after -f.
sources: Source paths added to the archive, in order.
options: Optional create settings. When omitted, the command uses TarCreateOptions().

Returns

SafeCmd: Safe command wrapper for the curated tar program and generated argv.

function builders.tar_extract

def tar_extract(
    archive: str | Path,
    *,
    destination: str | Path | None = None,
    allow_relative: bool = False,
) -> SafeCmd

Build a tar extract command.

Parameters

archive: Archive path passed after -f.
destination: Optional extraction directory passed after -C.
allow_relative: Allow archive and destination to be relative paths.

Returns

SafeCmd: Safe command wrapper for the curated tar program and generated argv.