Usage

A first pipeline

A pipeline starts from some data, gets steps appended to it, and is read as an async iterator:

import asyncio
from action0.pipeline import Pipeline


def double(value: int) -> list[int]:
    return [value * 2]


async def main() -> None:
    async with Pipeline(range(5)).add_step(double) as pipeline:
        print(sorted([item async for item in pipeline]))  # [0, 2, 4, 6, 8]


asyncio.run(main())

The results are sorted here for a reason: items come out in the order they finish, not the order they went in. As soon as a step has more than one worker, a quick item overtakes a slow one. Sort at the end, or carry a key on your items, if the original order matters.

What a step returns

A transformer takes one item and returns the items it produced, so the same signature covers mapping, fan-out and filtering:

def letters(word: str) -> list[str]:
    return list(word)  # fan out: one word becomes several letters


def drop_vowels(letter: str) -> list[str]:
    return [] if letter in "aeiou" else [letter]  # filter: emit nothing


pipeline = Pipeline(["hello", "world"]).add_step(letters).add_step(drop_vowels)
async with pipeline as running:
    print(sorted([item async for item in running]))
    # ['d', 'h', 'l', 'l', 'l', 'r', 'w']

The type follows the chain, so the checkers know what each step is handed:

def to_text(value: int) -> list[str]: ...
def to_length(text: str) -> list[int]: ...


pipeline = Pipeline([1, 22, 333]).add_step(to_text).add_step(to_length)
# Pipeline[int] -> Pipeline[str] -> Pipeline[int]
async with pipeline as running:
    print(sorted([item async for item in running]))  # [2, 3, 4]

Blocking work and async work

Every step has two independent knobs:

• async_workers — how many asyncio tasks consume the step’s input.

• thread_pool_size — how many threads those tasks may hand blocking calls to. This is what makes a plain blocking function usable in an event loop. Set it to 0 for a transformer that is already non-blocking.

async def and async generator transformers always run in the event loop, and ignore thread_pool_size. Everything else is dispatched to the step’s own pool.

def blocking(value: int) -> list[int]:
    time.sleep(0.1)  # a blocking HTTP call, a database query, ...
    return [value]


async def waiting(value: int) -> AsyncIterator[int]:
    await asyncio.sleep(0.1)  # already non-blocking
    yield value


pipeline = (
    Pipeline(range(12))
    .add_step(blocking, async_workers=4, thread_pool_size=4)
    .add_step(waiting, async_workers=12)
)
async with pipeline as running:
    items = [item async for item in running]
# 12 items in 0.4s — done one at a time this would take 2.4s

Both steps are resident the whole time: while waiting handles the first item, blocking is already working on the fifth.

A synchronous source blocks the loop just as a synchronous step does. Pass initial_in_thread=True when advancing it does real work — a database cursor, a file, a paging HTTP client:

Pipeline(rows_from_database(), initial_in_thread=True)

Backpressure

Buffers are unbounded by default, so a fast source will happily read everything into memory. Bound them with maxsize to make the producer wait instead:

pipeline = Pipeline(huge_source, maxsize=100).add_step(
    handle, async_workers=10, thread_pool_size=10, maxsize=100
)

With a bounded buffer, a step’s workers wait in put() while it is full, which stalls them, which stops them reading their own input — the pressure travels all the way back to the source.

Errors

Each step decides for itself what happens when its transformer raises.

def risky(value: int) -> list[int]:
    if value == 2:
        raise ValueError(f"cannot handle {value}")
    return [value]

ErrorPolicy.RAISE (the default) tears the whole pipeline down and re-raises the original exception at the consumer — not an ExceptionGroup:

try:
    async with Pipeline(range(5)).add_step(risky) as running:
        [item async for item in running]
except ValueError as exc:
    print(exc)  # cannot handle 2

ErrorPolicy.SKIP logs a warning and drops the item:

pipeline = Pipeline(range(5)).add_step(risky, on_error=ErrorPolicy.SKIP)
async with pipeline as running:
    print(sorted([item async for item in running]))  # [0, 1, 3, 4]

A callback gets the exception and the item, and returns replacement items — or None to drop it:

def replace(exc: Exception, item: int) -> list[int]:
    return [-1]


pipeline = Pipeline(range(5)).add_step(risky, on_error=replace)
async with pipeline as running:
    print(sorted([item async for item in running]))  # [-1, 0, 1, 3, 4]

Lifecycle

async with is the form to reach for: it guarantees the workers are stopped and the thread pools closed even when the consumer breaks early or raises.

async with pipeline as running:
    async for item in running:
        if done_enough(item):
            break  # the rest of the pipeline is torn down here

Iterating the pipeline directly works too, and cleans up once the last item has been read — but an early break then leaves the cleanup to the garbage collector, so prefer the context manager. await pipeline.aclose() does it by hand; it is idempotent, and safe on a pipeline that was never started.

A pipeline is single-use: its buffers are shut down when it finishes, so reading it a second time yields nothing. add_step returns a new pipeline rather than mutating the one it was called on, so a partially built pipeline can be shared as a base.

Watching it work

buffer_sizes() reports what is queued at each stage boundary, source buffer first — the quickest way to find the step everything is waiting for:

async with pipeline as running:
    await asyncio.sleep(0.01)
    print(running.buffer_sizes())  # [98, 0] — the source is way ahead

Custom buffers

Anything satisfying the Buffer protocol can sit between two steps; the pipeline creates asyncio.Queue objects only because a queue already satisfies it. Implement the four methods to buffer somewhere else — but reproduce the queue’s shutdown semantics, because that is how the pipeline detects a stage has run dry:

class MyBuffer[Item]:
    def qsize(self) -> int: ...
    async def put(self, item: Item) -> None: ...
    async def get(self) -> Item: ...
    def shutdown(self, immediate: bool = False) -> None: ...


pipeline = Pipeline(source).add_step(handle, buffer=MyBuffer())