API reference

Everything public is importable from the package root:

from action0.pipeline import Buffer, ErrorPolicy, Pipeline, StepDefinition

Pipeline

An asyncio pipeline: a chain of transformer steps that all run at the same time, each with its own number of concurrent workers and its own thread pool.

type action0.pipeline.pipeline.Produced = Iterable | AsyncIterable

What a transformer returns: any number of output items for the one input item.

type action0.pipeline.pipeline.Transformer = Callable[[In], Produced | Awaitable[Produced]]

A step’s work function. Takes one item and returns the items it produced, so a step can fan out (several items), map (exactly one) or filter (none). It may be a plain function, an async def function or an async generator function.

type action0.pipeline.pipeline.Source = Iterable | AsyncIterable

Items to read, synchronously or asynchronously.

type action0.pipeline.pipeline.InitialData = Source | Callable[[], Source | Awaitable[Source]]

The data a pipeline starts from: an (async) iterable, or a callable returning one — a callable is only invoked when the pipeline starts, so it is the way to feed a pipeline that is built before there is a running event loop.

type action0.pipeline.pipeline.ErrorHandler = Callable[[Exception, In], Produced | None]

Per-step error callback: gets the exception and the item that caused it, and returns replacement items to emit instead (or None to drop the item).

class action0.pipeline.pipeline.ErrorPolicy(*values)

What a step does when its transformer raises.

RAISE = 'raise'

Tear the whole pipeline down and re-raise at the consumer. The default.

SKIP = 'skip'

Log a warning, drop the item and carry on with the next one.

type action0.pipeline.pipeline.ErrorPolicyLike = ErrorPolicy | Literal['raise', 'skip']

An ErrorPolicy, or the plain string spelling of one.

class action0.pipeline.pipeline.StepDefinition(transformer, buffer=None, maxsize=0, async_workers=1, thread_pool_size=1, on_error=ErrorPolicy.RAISE, name=None)

One step of a pipeline: what to run, how often, and where to put the results.

Created by Pipeline.add_step(); you only meet it when inspecting a pipeline you have already built.

Parameters:

• transformer (Callable[[TypeVar(In)], Iterable[TypeVar(Out)] | AsyncIterable[TypeVar(Out)] | Awaitable[Iterable[TypeVar(Out)] | AsyncIterable[TypeVar(Out)]]])

• buffer (Buffer[TypeVar(Out)] | None, default: None)

• maxsize (int, default: 0)

• async_workers (int, default: 1)

• thread_pool_size (int, default: 1)

• on_error (ErrorPolicy | Callable[[Exception, TypeVar(In)], Iterable[TypeVar(Out)] | AsyncIterable[TypeVar(Out)] | None], default: <ErrorPolicy.RAISE: 'raise'>)

• name (str | None, default: None)

transformer: Transformer

The work function, called once per input item.

buffer: Buffer | None = None

The step’s output buffer; a fresh asyncio.Queue when None.

maxsize: int = 0

Bound of the created output buffer, 0 for unbounded. Ignored if buffer is set.

async_workers: int = 1

How many concurrent asyncio tasks consume this step’s input.

thread_pool_size: int = 1

Size of this step’s thread pool, or 0 to run the transformer in the event loop.

on_error: ErrorPolicy | ErrorHandler = 'raise'

What to do when the transformer raises.

name: str | None = None

Optional label, used in task names and log messages.

class action0.pipeline.pipeline.Pipeline(initial, *, buffer=None, maxsize=0, initial_in_thread=False, name=None)

A chain of steps that processes items concurrently from end to end.

Build a pipeline from some initial data, append steps, then read the results as an async iterator. Every step is resident and working at the same time: while step 2 handles the first item, step 1 is already producing the second.

Each step gets its own concurrency settings. async_workers is how many asyncio tasks consume the step’s input buffer, and thread_pool_size is how many threads those tasks may hand blocking work to — which is what makes a blocking function usable without stalling the event loop. Use thread_pool_size=0 for a transformer that is already non-blocking.

>>> import asyncio
>>> def double(value: int) -> list[int]:
...     return [value * 2]
>>> async def main() -> list[int]:
...     async with Pipeline(range(5)).add_step(double, async_workers=3) as pipeline:
...         return sorted([item async for item in pipeline])
>>> asyncio.run(main())
[0, 2, 4, 6, 8]

Items come out in whatever order they finish, not in input order: as soon as a step has more than one worker, a quick item overtakes a slow one. Sort the results, or keep a key on the items, if you need the original order.

async with is the recommended form. It guarantees that the workers are torn down and the thread pools closed even when the consumer stops early or raises; plain async for over the pipeline works too and cleans up once the last item has been read.

A pipeline is single-use: reading it a second time yields nothing, because its buffers have been shut down.

Parameters:

• initial (Iterable[TypeVar(Item)] | AsyncIterable[TypeVar(Item)] | Callable[[], Iterable[TypeVar(Item)] | AsyncIterable[TypeVar(Item)] | Awaitable[Iterable[TypeVar(Item)] | AsyncIterable[TypeVar(Item)]]])

• buffer (Buffer[TypeVar(Item)] | None, default: None)

• maxsize (int, default: 0)

• initial_in_thread (bool, default: False)

• name (str | None, default: None)

• initial – the items the pipeline starts from, or a callable returning them (see InitialData).

• buffer – buffer for the initial items; a fresh asyncio.Queue is created when None.

• maxsize – bound of the created initial buffer, 0 for unbounded. Ignored when buffer is given.

• initial_in_thread – iterate a synchronous initial in a thread. Set this when advancing the source blocks (a database cursor, a file, a paging HTTP client) — otherwise it blocks the event loop, and with it every step of the pipeline.

• name – optional label, used in task names and log messages.

property name: str

The pipeline’s label.

property steps: tuple[StepDefinition[Any, Any], ...]

The steps appended so far, in order.

add_step(transformer, *, buffer=None, maxsize=0, async_workers=1, thread_pool_size=1, on_error=ErrorPolicy.RAISE, name=None)

Append a step that consumes this pipeline’s current output.

Returns a new pipeline parameterised by the new step’s output type, so the static type follows the chain as it is built. The original is left as it was.

Parameters:

• transformer (Callable[[TypeVar(Item)], Iterable[TypeVar(New)] | AsyncIterable[TypeVar(New)] | Awaitable[Iterable[TypeVar(New)] | AsyncIterable[TypeVar(New)]]]) – the work function, called once per item (see Transformer).

• buffer (Buffer[TypeVar(New)] | None, default: None) – the step’s output buffer; a fresh asyncio.Queue is created when None.

• maxsize (int, default: 0) – bound of the created output buffer, 0 for unbounded. Ignored when buffer is given. Bound it to apply backpressure to the steps upstream.

• async_workers (int, default: 1) – how many asyncio tasks run this step concurrently.

• thread_pool_size (int, default: 1) – size of the step’s own thread pool. A synchronous transformer is dispatched to it, so blocking calls do not stall the event loop; 0 runs the transformer in the loop instead. Ignored for async def and async generator transformers, which always run in the loop.

• on_error (ErrorPolicy | Literal['raise', 'skip'] | Callable[[Exception, TypeVar(Item)], Iterable[TypeVar(New)] | AsyncIterable[TypeVar(New)] | None], default: <ErrorPolicy.RAISE: 'raise'>) – an ErrorPolicy, or a callback returning replacement items (see ErrorHandler).

• name (str | None, default: None) – optional label, used in task names and log messages.

Return type:

Pipeline[TypeVar(New)]

Returns:

a new pipeline ending in this step.

buffer_sizes()

How much is queued up right now, one entry per buffer, source buffer first.

Useful for spotting the step that is holding everything else up.

Return type:

list[int]

Returns:

the buffered item counts, empty before the pipeline has started.

async start()

Create the buffers, thread pools and worker tasks and let them run.

Called automatically by async with and by the first iteration, so there is rarely a reason to call it yourself.

Raises:

RuntimeError – if the pipeline has already been started.

Return type:

None

async aclose()

Shut everything down: stop the workers, close the thread pools.

Safe to call more than once, and safe to call on a pipeline that was never started.

Return type:

None

async __aenter__()

Return type:

Self

async __aexit__(exc_type, exc, traceback)

Parameters:

• exc_type (type[BaseException] | None)

• exc (BaseException | None)

• traceback (TracebackType | None)

Return type:

None

__aiter__()

Return type:

AsyncIterator[TypeVar(Item)]

Buffer

The buffer that hands items from one pipeline step to the next.

class action0.pipeline.buffer.Buffer(*args, **kwargs)

The slice of asyncio.Queue a pipeline needs between two steps.

asyncio.Queue satisfies this protocol as it is, which is why the pipeline creates plain queues when no buffer is supplied. Implement the protocol yourself to buffer somewhere other than the event loop’s memory.

Implementations must reproduce the queue’s shutdown semantics, because the pipeline detects “this stage has run dry” from them and nothing else:

• after shutdown(), put() raises asyncio.QueueShutDown right away,

• get() raises asyncio.QueueShutDown once the items still buffered have been handed out — or immediately, when shut down with immediate=True.

A bounded buffer is what applies backpressure: while it is full, the producing step’s workers wait in put() instead of reading more input.

qsize()

Return type:

int

Returns:

the number of items currently buffered.

async put(item)

Append an item, waiting while the buffer is full.

Parameters:

item (TypeVar(Item)) – the item to buffer.

Raises:

asyncio.QueueShutDown – if the buffer has been shut down.

Return type:

None

async get()

Take the oldest buffered item, waiting while the buffer is empty.

Return type:

TypeVar(Item)

Returns:

the item.

Raises:

asyncio.QueueShutDown – if the buffer is shut down and drained.

shutdown(immediate=False)

Signal that no more items will be added.

Parameters:

immediate (bool, default: False) – also discard the items still buffered, so that waiting consumers stop right away instead of draining first.

Return type:

None