Pipelines and concurrency · API reference

class

`Pipeline`

class Pipeline(parts: 'tuple[SafeCmd, ...]')

A sequence of SafeCmd stages connected via stdout/stdin piping.

Attributes

parts tuple[SafeCmd, ...]: The stages in execution order. The first stage's stdin is the pipeline's stdin and the last stage's stdout is its stdout; each intervening stage feeds the next. Always at least _MIN_PIPELINE_STAGES long, as __post_init__ enforces.

6 members

field Pipeline.parts

parts: 'tuple[SafeCmd, ...]'

classmethod Pipeline.concat

def concat(
    left: 'SafeCmd | Pipeline',
    right: 'SafeCmd | Pipeline',
) -> 'Pipeline'

Compose a pipeline from two stage operands.

Parameters

left SafeCmd | Pipeline: A command or pipeline whose stages come first.
right SafeCmd | Pipeline: A command or pipeline whose stages follow left's.

Returns

Pipeline: A pipeline whose stages are *left*'s followed by *right*'s.

Raises

TypeError: If either operand is neither a SafeCmd nor a Pipeline. Structural matching cannot express exhaustiveness, so an unhandled operand would otherwise fall out of the match and return None from a function that promises a pipeline, deferring the failure to a distant attribute access.

method Pipeline.run

async def run(
    *,
    output: RunOutputOptions | None = None,
    timeout: float | None = None,
    context: ExecutionContext | None = None,
    **deprecated_flags: typ.Unpack[_DeprecatedOutputFlags],
) -> PipelineResult

Execute the pipeline asynchronously with streaming and backpressure.

Parameters

output RunOutputOptions | None, default=None: Capture and echo settings for every observed pipeline stream. The default bounds each echoed line to 64 KiB; None for max_echo_line_bytes restores unbounded mirroring without changing capture.
timeout float | None, default=None: Maximum pipeline execution time in seconds.
context ExecutionContext | None, default=None: Execution settings, including echo sinks and text encoding.
**deprecated_flags bool: Deprecated capture and echo keyword arguments. Do not combine them with output.

Returns

PipelineResult: The outcome for every stage and complete captured streams when capture is enabled.

Raises

ValueError: If output is combined with deprecated flags, or if it names a stdin/stdout/stderr target, which pipeline stages do not honour.
PermissionError: If a pipeline command is not allowed by the active scope.
TimeoutError: If execution exceeds the effective timeout.

method Pipeline.run_sync

def run_sync(
    *,
    output: RunOutputOptions | None = None,
    timeout: float | None = None,
    context: ExecutionContext | None = None,
    **deprecated_flags: typ.Unpack[_DeprecatedOutputFlags],
) -> PipelineResult

Execute the pipeline synchronously via asyncio.run.

Parameters

output RunOutputOptions | None, default=None: Capture and echo settings. The 64 KiB default bounds mirrored lines; max_echo_line_bytes=None restores unbounded echoing while leaving captured output complete.
timeout float | None, default=None: Maximum pipeline execution time in seconds.
context ExecutionContext | None, default=None: Execution settings, including echo sinks and text encoding.
**deprecated_flags bool: Deprecated capture and echo keyword arguments. Do not combine them with output.

Returns

PipelineResult: The outcome for every stage and complete captured streams when capture is enabled.

Raises

ValueError: If output is combined with deprecated flags, or if it names a stdin/stdout/stderr target, which pipeline stages do not honour.
PermissionError: If a pipeline command is not allowed by the active scope.
TimeoutError: If execution exceeds the effective timeout.

method Pipeline.run_bytes

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

Execute the pipeline, capturing its output as bytes.

Every stage runs exactly as it does under run(); the difference is what the parent keeps. Each stage's captured stderr and the final stage's captured stdout are reported as the child wrote them, which is what lets a pipeline relay a binary payload end to end. An interior stage's stdout is consumed by the next stage rather than captured, so it is reported as None exactly as it is in text mode.

The pipeline's own stdin feeding is unchanged: the first stage reads from the parent process's stdin, as it always does, and no stage's output is interpreted on the way through.

Parameters

output RunOutputOptions | None, default=None: Capture and echo settings for every observed pipeline stream. on_line is rejected: bytes mode returns each stream'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 pipeline execution time in seconds.
context ExecutionContext | None, default=None: Execution settings, including echo sinks and text encoding.

Returns

BytesPipelineResult: Per-stage outcomes plus the final stage's captured stdout, as bytes wherever a stream was captured.

Raises

ValueError: If output.on_line is set.
PermissionError: If a pipeline command is not allowed by the active scope.
TimeoutError: If execution exceeds the effective timeout.

method Pipeline.run_bytes_sync

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

Execute the pipeline synchronously, capturing its output as bytes.

Parameters

output RunOutputOptions | None, default=None: Capture and echo settings. on_line is rejected: bytes mode returns each stream'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 pipeline execution time in seconds.
context ExecutionContext | None, default=None: Execution settings, including echo sinks and text encoding.

Returns

BytesPipelineResult: Per-stage outcomes plus the final stage's captured stdout, as bytes wherever a stream was captured.

Raises

ValueError: If output.on_line is set.
PermissionError: If a pipeline command is not allowed by the active scope.
TimeoutError: If execution exceeds the effective timeout.

class

`PipelineResult`

cuprum/sh/results.py

class PipelineResult(
    stages: tuple[CommandResult, ...],
    failure_index: int | None = None,
)

Structured result returned by pipeline execution.

Attributes

stages: Command results for each pipeline stage, in execution order. For stages whose stdout is streamed into the next stage, stdout is None. The final stage carries captured stdout when enabled.
failure_index: Index of the stage that triggered fail-fast termination, or None when all stages completed successfully.

6 members

field PipelineResult.stages

stages: tuple[CommandResult, ...]

field PipelineResult.failure_index

failure_index: int | None

property PipelineResult.final

final: CommandResult

The result from the final pipeline stage.

Returns

CommandResult: The last stage's result in execution order.

property PipelineResult.failure

failure: CommandResult | None

The stage that triggered fail-fast termination, if any.

Returns

CommandResult | None: The failing stage result, or None when no stage triggered fail-fast termination.

property PipelineResult.ok

ok: bool

Whether every pipeline stage exited successfully.

Returns

bool: True when every stage result is successful; otherwise False.

property PipelineResult.stdout

stdout: str | None

Captured output from the final pipeline stage.

Returns

str | None: The final stage's captured standard output, or None when capture was disabled.

class

`BytesPipelineResult`

cuprum/sh/results.py

class BytesPipelineResult(
    stages: tuple[BytesCommandResult, ...],
    failure_index: int | None = None,
)

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

Attributes

stages: Per-stage results in execution order. Stages whose stdout is streamed into the next stage carry stdout of None; the final stage carries captured stdout bytes when capture is enabled.
failure_index: Index of the stage that triggered fail-fast termination, or None when all stages completed successfully.

6 members

field BytesPipelineResult.stages

stages: tuple[BytesCommandResult, ...]

field BytesPipelineResult.failure_index

failure_index: int | None

property BytesPipelineResult.final

final: BytesCommandResult

The result from the final pipeline stage.

Returns

BytesCommandResult: The last stage's result in execution order.

property BytesPipelineResult.failure

failure: BytesCommandResult | None

The stage that triggered fail-fast termination, if any.

Returns

BytesCommandResult | None: The failing stage result, or None when no stage triggered fail-fast termination.

property BytesPipelineResult.ok

ok: bool

Whether every pipeline stage exited successfully.

Returns

bool: True when every stage result is successful; otherwise False.

property BytesPipelineResult.stdout

stdout: bytes | None

Captured output from the final pipeline stage.

Returns

bytes | None: The final stage's captured standard output bytes, or None when capture was disabled.

function

`run_concurrent`

cuprum/concurrent.py

async def run_concurrent(
    *commands: SafeCmd,
    config: ConcurrentConfig | None = None,
) -> ConcurrentResult

Execute multiple SafeCmd instances concurrently.

Parameters

*commands: SafeCmd instances to execute concurrently.
config: Configuration for concurrent execution. When None, uses default ConcurrentConfig() which allows unlimited concurrency, captures output, does not echo, uses current context, and collects all results (no fail-fast).

Returns

ConcurrentResult: Aggregated results in submission order.

Raises

ValueError: If config.concurrency < 1 or no commands provided.
ForbiddenProgramError: If current_context().check_allowed() rejects a command before task creation.

function

`run_concurrent_sync`

cuprum/concurrent.py

def run_concurrent_sync(
    *commands: SafeCmd,
    config: ConcurrentConfig | None = None,
) -> ConcurrentResult

Execute multiple SafeCmd instances concurrently (synchronous wrapper).

This method mirrors run_concurrent() by driving the event loop internally. All parameters and return semantics are identical.

Parameters

*commands: SafeCmd instances to execute concurrently.
config: Configuration for concurrent execution. When None, uses default ConcurrentConfig() which allows unlimited concurrency, captures output, does not echo, uses current context, and collects all results (no fail-fast).

Returns

ConcurrentResult: Aggregated results in submission order.

class

`ConcurrentConfig`

cuprum/_concurrent_config.py

class ConcurrentConfig(
    concurrency: int | None = None,
    capture: bool = True,
    echo: bool = False,
    echo_stdout: bool | None = dc.field(default=None, kw_only=True),
    echo_stderr: bool | None = dc.field(default=None, kw_only=True),
    context: ExecutionContext | None = None,
    fail_fast: bool = False,
)

Configuration for concurrent command execution.

Attributes

concurrency: Maximum number of commands to execute concurrently. When None, all commands are started immediately with no throttling. Must be >= 1 when provided. A non-integer or bool value raises TypeError; a value below 1 raises ValueError.
capture: When True, capture stdout/stderr for each command into the CommandResult. When False, output is not captured and stdout will be None in results.
echo: When True, tee output to configured sinks (stdout/stderr by default) in addition to capturing.
echo_stdout: When True, tee only stdout to its configured sink; when False, stdout is never echoed. When None (default), inherits echo. Keyword-only, so constructing this dataclass positionally keeps binding context and fail_fast as before.
echo_stderr: When True, tee only stderr to its configured sink; when False, stderr is never echoed. When None (default), inherits echo. Keyword-only, so constructing this dataclass positionally keeps binding context and fail_fast as before.
context: Shared execution context for all commands. When None, uses the current context from the execution environment.
fail_fast: When True, cancel remaining commands after the first command exits with a non-zero status. When False (default), all commands run to completion regardless of individual failures.

7 members

field ConcurrentConfig.concurrency

concurrency: int | None

field ConcurrentConfig.capture

capture: bool

field ConcurrentConfig.echo

echo: bool

field ConcurrentConfig.echo_stdout

echo_stdout: bool | None

field ConcurrentConfig.echo_stderr

echo_stderr: bool | None

field ConcurrentConfig.context

context: ExecutionContext | None

field ConcurrentConfig.fail_fast

fail_fast: bool

class

`ConcurrentResult`

cuprum/_concurrent_config.py

class ConcurrentResult(
    results: tuple[CommandResult, ...],
    failures: tuple[int, ...] = (),
    submission_indices: tuple[int, ...] | None = None,
) -> None

Aggregated result from concurrent command execution.

Attributes

results: CommandResult instances in submission order (not completion order). When fail_fast=True is used, execution may cancel remaining commands after the first failure; in this case, results contains only the completed (non-cancelled) CommandResult instances and may be a subset of the originally submitted commands.
failures: Indices into results for commands that exited non-zero, in ascending order. These are positions within the (possibly compacted) results tuple, **not** original submission positions; in fail-fast mode results omits cancelled commands, so a failures index does not equal the submitted command's position. Use failure_submission_indices for the submission-stable mapping.
submission_indices: Original submission index of each entry in results, parallel to it. In collect-all mode this is the identity (0, 1, …, n-1); in fail-fast mode it lists the submission positions of the completed commands. Lets callers map any result — or failure — back to the command they submitted, uniformly across both execution modes. Pass None (the default) to backfill the identity sequence (even for empty results); any supplied sequence — including an empty tuple — whose length differs from results raises ValueError.

6 members

field ConcurrentResult.results

results: tuple[CommandResult, ...]

field ConcurrentResult.failures

failures: tuple[int, ...]

field ConcurrentResult.submission_indices

submission_indices: tuple[int, ...]

property ConcurrentResult.ok

ok: bool

Whether every command completed successfully.

Returns

bool: True when no result is recorded as a failure; otherwise False.

property ConcurrentResult.first_failure

first_failure: CommandResult | None

The earliest failed command result, if any.

Returns

CommandResult | None: The result at the first recorded failure index, or None when every command succeeded.

property ConcurrentResult.failure_submission_indices

failure_submission_indices: tuple[int, ...]

Original submission indices of failed commands, ascending.

Unlike failures (positions within the compacted results), these are stable across both collect-all and fail-fast modes: each value is the position at which the failing command was submitted.