Pipeline(iterable=(), default=None, name=None, *, run_now=False, run_args=(), run_kwargs=None)

A simple asynchronous pipeline for executing callable steps sequentially.

Each step is represented by a tuple containing a callable and optional positional and keyword arguments:

(function,)
(function, args)
(function, kwargs)
(function, args, kwargs)

Where args may be a tuple or mapping when used alone, but must be a tuple when kwargs is provided.
The pipeline always passes the result of the previous step as the first positional argument to the next step.
The pipeline is snapshotted when execution starts, so modifications to the pipeline after run() has started do not affect the current execution.

Example:

def add(value, amount):
    return value + amount

pipeline = Pipeline([
    (add, (5,)),
    (add, (10,)),
)
pipeline.run(0).wait().result  # 15

Args:
iterable: An iterable containing pipeline steps.
Defaults to an empty pipeline.
default: The value used for the first step when run() is called without an explicit default.
Defaults to None.
name: The name assigned to the worker thread and pipeline.
Must be a non-empty string or None.
Defaults to None.
run_now: Whether to start the pipeline immediately after initialization.
Defaults to False.
run_args: A tuple of positional arguments passed to run() when run_now is enabled.
Defaults to ().
run_kwargs: A mapping of keyword arguments passed to run() when run_now is enabled.
Defaults to None.

Raises:
TypeError: If any item in iterable is not a valid step, if name is not a string or None, if run_args is not a tuple, or if run_kwargs is not a mapping.
ValueError: If name is an empty string.

Execution

run(default=_DEFAULT, delay=0, daemon=False, stop_on_error=True, handler_error=None)

Run the pipeline asynchronously in a worker thread.

The supplied default value is stored as the initial result.
Each step receives the result of the preceding step as its first positional argument.

The pipeline is snapshotted when execution starts, so modifications to the configured steps do not affect the current execution.
Execution stops when all steps have completed, stop() is called, or an exception is raised while stop_on_error is enabled.

Args:
default: The initial value passed to the first step.
If omitted, the pipeline's default value is used.
None may be passed explicitly.
delay: Delay in seconds before the first step and between subsequent steps.
Must be a non-negative integer or floating-point number.
Defaults to 0.
daemon: Whether the worker thread should be a daemon thread.
Defaults to False.
stop_on_error: Whether execution should stop after the first exception.
When disabled, the failed step's result is not added to results, the exception is added to errors, and execution continues with the previous result.
Defaults to True.
handler_error: An optional callable invoked when a step raises an exception.
It receives an ErrorContext instance containing the one-based step index, the failed step, the exception object, and the result passed to the failed step.
The handler's own exceptions are not caught by the pipeline.
Defaults to None.

Returns:
This pipeline instance.

Raises:
TypeError: If delay is not an integer or floating-point number, or if handler_error is not callable.
ValueError: If delay is negative.
RuntimeError: If the pipeline is already running.

__call__(*args, **kwargs)

Run the pipeline.

Arguments are passed directly to run().

Returns:
This pipeline instance.

wait(timeout=None)

Wait until the currently running pipeline finishes.

This method blocks only while the pipeline is running.
If timeout expires before the pipeline finishes, the worker thread continues running in the background.

Args:
timeout: Maximum time to wait in seconds, or None to wait indefinitely.
Must be non-negative or None.
Defaults to None.

Returns:
This pipeline instance.

Raises:
TypeError: If timeout is not an integer, a floating-point number, or None.
ValueError: If timeout is negative.

stop()

Request the running pipeline to stop and wait for termination.

Returns:
The one-based index of the step at which execution was stopped, or 0 if the pipeline was not running.

skip()

Request the currently running pipeline to skip its next step.

If the worker has not yet checked the skip event, the next step that reaches the worker's skip check will be skipped.

Returns:
This pipeline instance.

rerun(*args, **kwargs)

Stop the current execution and start the pipeline again.

All arguments are passed directly to run().

Returns:
This pipeline instance.

Manual Step Execution

run_step(index, default=None)

Execute a single pipeline step synchronously.

This method does not create or modify the worker thread and does not store the result or exception in the pipeline's results or errors stacks.
Exceptions are allowed to propagate to the caller.

Args:
index: One-based index of the pipeline step to execute.
default: Value passed to the selected step as its first argument.
Defaults to None.

Returns:
The value returned by the selected step.

Raises:
TypeError: If index is not an integer.
ValueError: If index is less than 1.
IndexError: If index is outside the pipeline.
RuntimeError: If the pipeline is currently running.

execute(step, default=None)

Execute a pipeline step synchronously.

This method validates and executes the specified step directly.
It does not create or modify the worker thread and does not store the result or exception in the pipeline's results or errors stacks.
Exceptions are allowed to propagate to the caller.

Args:
step: The pipeline step to execute.
default: Value passed to the step as its first argument.
Defaults to None.

Returns:
The value returned by the step.

Raises:
TypeError: If step has an invalid format.
RuntimeError: If the pipeline is currently running.

Execution State

running property

Return whether the pipeline currently has a running worker thread.

step property

Return the one-based index of the step currently running, or 0 when idle.

calls property

Return the stack containing successfully executed pipeline steps.

The stack records each step after its execution completes successfully.
Therefore, a step is not included in calls if it raises an exception.

Skipped steps are not added to the stack.
Steps that have not yet been reached are not included.

Returns:
The stack containing the steps successfully executed during the current or most recent execution.

results property

Return the stack containing initial and successful step results.

errors property

Return the stack containing exceptions raised by pipeline steps.

result property

Return the most recent result of the pipeline.

The initial value is considered the result when no pipeline step has successfully completed.

Returns:
The initial value or the result produced by the last successfully executed step.
None if the pipeline has not been run yet.

Raises:
RuntimeError: If the pipeline is currently running.

error(reraise=True)

Return or raise the most recent exception raised by the pipeline.

By default, the exception is raised.
If reraise is False, the exception is returned instead.

Args:
reraise: Whether to raise the most recent exception.
Defaults to True.

Returns:
The most recent exception if reraise is False.
None if no exception was raised.

Raises:
RuntimeError: If the pipeline is currently running.
Exception: The most recent exception raised by a pipeline step when reraise is True.

Step Management

add(step)

Append a validated step to the pipeline.

Args:
step: The step to append.

Raises:
TypeError: If step has an invalid format.

insert(index, step)

Insert a validated step at the specified index.

Args:
index: One-based position at which to insert the step.
step: The step to insert.

Raises:
TypeError: If index is not an integer or step is invalid.
IndexError: If index is outside the pipeline.

update(iterable)

Append multiple validated steps to the pipeline.

All steps are validated before being added, so the pipeline is left unchanged if any step is invalid.

Args:
iterable: An iterable of pipeline steps.

Raises:
TypeError: If any item in iterable has an invalid step format.

remove(step)

Remove the first matching step from the pipeline.

Args:
step: The pipeline step to remove.

Raises:
TypeError: If step has an invalid format.
ValueError: If step is not found in the pipeline.

discard(step)

Remove the first matching step from the pipeline if present.

If the step is not found, the pipeline is left unchanged.

Args:
step: The pipeline step to remove.

Raises:
TypeError: If step has an invalid format.

pop(index)

Remove and return the step at the specified index.

Args:
index: One-based index of the step to remove.

Returns:
The removed pipeline step.

Raises:
TypeError: If index is not an integer.
IndexError: If index is outside the pipeline.

clear()

Remove all steps from the pipeline.

reverse()

Reverse the order of pipeline steps in place.

index(step)

Return the one-based index of the first matching pipeline step.

Args:
step: The pipeline step to search for.

Returns:
The one-based index of the first matching step.

Raises:
TypeError: If step has an invalid format.
ValueError: If step is not found in the pipeline.

count(step)

Return the number of occurrences of a pipeline step.

Args:
step: The pipeline step to count.

Returns:
The number of times step occurs in the pipeline.

Raises:
TypeError: If step has an invalid format.

Item Access

__getitem__(index)

Return the pipeline step at the specified index.

Args:
index: One-based index of the pipeline step to retrieve.

Returns:
The pipeline step at the specified index.

Raises:
TypeError: If index is not an integer.
IndexError: If index is outside the pipeline.

__setitem__(index, step)

Replace the pipeline step at the specified index.

Args:
index: One-based index of the pipeline step to replace.
step: The new pipeline step.

Raises:
TypeError: If index is not an integer or step is invalid.
IndexError: If index is outside the pipeline.

__delitem__(index)

Remove the pipeline step at the specified index.

Args:
index: One-based index of the pipeline step to remove.

Raises:
TypeError: If index is not an integer.
IndexError: If index is outside the pipeline.

Configuration

default property writable

Return the default value used when run() receives no initial value.

name property writable

Return the name of the worker thread and pipeline.

steps property

Return a deep copy of the configured pipeline steps.

Copying

copy()

Return a shallow copy of the pipeline.

The pipeline configuration, default value, and name are shallow-copied.
Execution state, including results, errors, the current step, and the worker thread, is not copied.

Returns:
A new pipeline instance with a shallow copy of the configuration.

Raises:
RuntimeError: If the pipeline is currently running.

__copy__()

Create a shallow copy of the pipeline.

Only the pipeline configuration, default value, and name are copied.
Execution state, including results, errors, the current step, and the worker thread, is reset in the new pipeline.

Returns:
A new pipeline instance with a shallow copy of the configuration.

Raises:
RuntimeError: If the pipeline is currently running.

__deepcopy__(memo)

Create a deep copy of the pipeline.

The pipeline configuration, default value, and name are deep-copied.
Execution state, including results, errors, the current step, and the worker thread, is reset in the new pipeline.

Returns:
A new pipeline instance with a deep copy of the configuration.

Raises:
RuntimeError: If the pipeline is currently running.

Context Management

__enter__()

Enter the context manager and start the pipeline if it is not running.

Returns:
This pipeline instance.

__exit__(exc_type, exc_value, traceback)

Exit the context manager and stop the pipeline.

Iteration

__iter__()

Iterate over the configured pipeline steps.

Yields:
Each pipeline step in its configured order.

__reversed__()

Return an iterator that yields pipeline steps in reverse order.

Returns:
An iterator over the pipeline steps in reverse order.

Protocols

__bool__()

Return whether the pipeline contains any configured steps.

Returns:
True if the pipeline contains at least one step, otherwise False.

__len__()

Return the number of configured pipeline steps.

__contains__(item)

Return whether a callable or step configuration exists in the pipeline.

Callable items are matched by identity.
Step items must be valid step configurations and are matched against the full configuration.

Args:
item: A callable or a step tuple to search for.

Returns:
True if the callable or valid step configuration is present in the pipeline, otherwise False.

Raises:
TypeError: If item is not callable or a tuple, or if a tuple is not a valid step.

String Representation

__str__()

Return a human-readable representation of the pipeline.

__repr__()

Return the developer-oriented representation of the pipeline.

Operators

__iadd__(step)

Append a pipeline step in place.

This is equivalent to calling add() with the specified step.

Args:
step: The pipeline step to append.

Returns:
This pipeline instance.

Raises:
TypeError: If step has an invalid format.

__add__(other)

Return a new pipeline by concatenating pipeline steps.

The original pipeline and the iterable are not modified.

Args:
other: An iterable containing pipeline steps to append.

Returns:
A new pipeline containing the steps from this pipeline followed by the steps from other.

Raises:
TypeError: If any item in other has an invalid step format.

__radd__(other)

Return a new pipeline by prepending iterable steps.

The original pipeline and the iterable are not modified.

Args:
other: An iterable containing pipeline steps to prepend.

Returns:
A new pipeline containing the steps from other followed by the steps from this pipeline.

Raises:
TypeError: If any item in other has an invalid step format.

__mul__(count)

Return a new pipeline with its steps repeated.

The original pipeline is not modified.

Args:
count: Number of times to repeat the pipeline steps.
Must be a positive integer. Non-integer values are unsupported.

Returns:
A new pipeline containing the repeated steps.

Raises:
ValueError: If count is less than one.

Comparison

__eq__(other)

Return whether two pipelines have equal configurations.

Two pipelines are considered equal when they have the same pipeline steps, default value, and name.
Execution state is not considered.

Args:
other: The object to compare with.

Returns:
True if other is a pipeline with the same steps, default value, and name, otherwise False.