async-conveyor

Run a tree of processing stages concurrently on anyio, with bounded queues between them. You write the stages; the conveyor wires the queues, runs the workers, applies backpressure, closes resources and reports what happened.

from async_conveyor import Conveyor

flow = (
    Crawler(base_url="https://example.com").with_settings(buffer_size=20)
    | Parser().with_settings(workers=4)
    | (
        (Embedder(model="tiny").with_settings(workers=3, batch_size=8) | VectorStore())
        & (WordCounter() | ReportWriter())
    )
)

await Conveyor(flow).run()

• a | b feeds b from a; a | b & c feeds both b and c (broadcast). One root, no fan-in.
• Sync stages run in threads, async stages on the event loop, batch stages get lists.
• mypy checks every edge: a stage that emits Page can only be followed by one that takes Page.
• Bounded queues: a slow stage blocks its producer instead of filling memory.
• One exception anywhere stops everything, closes every resource and surfaces unwrapped from run().
• Telemetry hooks for metrics, tracing and errors; a logging default and an in-memory summary are included.

Python 3.11+, anyio is the only dependency.

uv add async-conveyor   # or: pip install async-conveyor

Stages

A stage is a class with a process method. Pick a base by how process runs and what it receives:

process is a generator: yield sends an item downstream, as many or as few as you like. The three type parameters are the input type, the output type and the type of the stage's resources.

import re
from collections.abc import AsyncGenerator, Generator
from dataclasses import dataclass

import httpx

from async_conveyor import AsyncStage, SyncStage


@dataclass(eq=False, frozen=True)
class Crawler(AsyncStage[None, Page, httpx.AsyncClient]):
    base_url: str

    def stage_context(self) -> httpx.AsyncClient:
        return httpx.AsyncClient(base_url=self.base_url)  # an async context manager: entered once per run

    async def process(self, item: None, client: httpx.AsyncClient) -> AsyncGenerator[Page, None]:
        for path in ("/a", "/b", "/c"):
            response = await client.get(path)
            yield Page(url=str(response.url), html=response.text)


class Parser(SyncStage[Page, Chunk, re.Pattern[str]]):
    def stage_context(self) -> AbstractAsyncContextManager[re.Pattern[str]]:
        return nullcontext(re.compile(r"\w+"))  # plain state, nothing to close

    def process(self, item: Page, words: re.Pattern[str]) -> Generator[Chunk, None, None]:
        for index, word in enumerate(words.findall(item.html)):
            yield Chunk(url=item.url, index=index, text=word)


class Store(SyncStage[Chunk, None, Db]):
    @asynccontextmanager
    async def stage_context(self) -> AsyncIterator[Db]:
        with Db.connect() as db:  # a sync client, used from the stage's threads
            yield db

    def process(self, item: Chunk, db: Db) -> Generator[None, None, None]:
        db.insert(item)
        return
        yield  # a terminal stage: a generator that never yields

Things to know:

• Configuration goes through the constructor. A stage is an ordinary class, or a dataclass. Dataclasses need eq=False so that instances stay usable as topology keys; frozen=True is fine, a stage holds no run state.
• The root is called once with None. The end of its generator is the end of the stream.
• stage_context() holds the stage's resources: HTTP clients, DB pools, loaded models. It is entered once on the event loop before the stage's workers start and exited after they finish, also on error or cancellation. Its value is passed to every process call. Without it, process gets None. The value must be safe for workers concurrent calls; for a sync stage it is used from threads.
• Items are immutable by contract. On broadcast the same object goes to every child.
• Retries and timeouts belong inside process. An exception escaping it stops the conveyor.
• A stage's name defaults to its class name. It appears in telemetry and errors and must be unique within a topology; set it when one class appears twice (name: str = "..." as a dataclass field, or a class attribute).

Flows

Compose stages with | and &. The result type-checks under mypy: Crawler | Parser is accepted only if Crawler emits what Parser takes.

flow = (
    Crawler(base_url=url)
    | Parser().with_settings(workers=4, buffer_size=10)
    | (
        (Embedder(model="m").with_settings(workers=3, batch_size=8) | Store())
        & (WordCounter() | Report())
    )
)

Rules:

• a | b: b consumes what a emits. a | b & c: b and c both consume what a emits.
• Nothing may follow a fan-out: (b & c) | d would be fan-in. It is a type error and a TypeError.
• & binds tighter than | in Python, so wrap every branch longer than one stage in parentheses, and write a fan-out as its own parenthesised block with one branch per line, as above. Forgetting parentheses is loud, never a silently different topology: a | b & c | d fails because it parses as (a | (b & c)) | d.
• Every position needs its own stage instance. The same instance may appear in several different flows.

If you prefer data over operators, Conveyor also takes two dicts: stage instance to children, and stage instance to StageSettings.

crawler, parser, store = Crawler(base_url=url), Parser(), Store()
Conveyor({crawler: [parser], parser: [store]}, {parser: StageSettings(workers=4, buffer_size=10)})

Settings

with_settings(...) attaches runtime settings to a stage; a bare stage gets the defaults. These are the numbers you would load from a config file.

Two knobs, two kinds of memory:

• workers bounds how many items are being processed at once. If processing is what costs memory (opening images, model inference), this is the limit that matters.
• buffer_size bounds how many items wait in a queue. If the items themselves are large (images passed between stages), this is the limit that matters.

A full queue blocks the producer, and the block propagates upstream to the root. The slowest branch of a fan-out sets the pace for everything above it.

Batching

A batch stage receives a list. A batch starts with the first item that arrives and is handed to process when it holds batch_size items or batch_timeout seconds have passed since that first item; the end of the stream flushes a partial batch. Every worker fills its own batch. The parent's buffer_size should be at least batch_size, or batches fill only as fast as the parent produces.

@dataclass(eq=False, frozen=True)
class Embedder(AsyncBatchStage[Chunk, Vector, Model]):
    model: str

    def stage_context(self) -> AbstractAsyncContextManager[Model]:
        return nullcontext(load(self.model))

    async def process(self, items: list[Chunk], model: Model) -> AsyncGenerator[Vector, None]:
        for chunk, values in zip(items, await model.embed([c.text for c in items]), strict=True):
            yield Vector(chunk=chunk, values=values)

Telemetry

Conveyor(flow, telemetry=...) takes anything that implements the Telemetry protocol:

class Telemetry(Protocol):
    def item_span(self, stage_name: str, worker: int) -> AbstractAsyncContextManager[None]: ...  # wraps a call
    async def on_item_done(self, stats: ItemStats) -> None: ...  # after every process() call
    async def on_error(self, stage_name: str, worker: int, exc: BaseException, item: object) -> None: ...

ItemStats carries the stage name, worker id, how many inputs the call consumed, how many items it emitted, how long it took, how long it waited on full output queues, and a snapshot of the stage's queues and busy workers. Open an OpenTelemetry span in item_span, push the stats to Prometheus in on_item_done, report to Sentry in on_error.

Two implementations are included:

• LoggingTelemetry (the default) logs every item at DEBUG and every error at ERROR on the async_conveyor.telemetry logger, with structured extra fields.
• SummaryTelemetry aggregates per-stage totals in memory, for scripts, notebooks and tests:

telemetry = SummaryTelemetry()
await Conveyor(flow, telemetry=telemetry).run()
print(telemetry.stages["Embedder"].blocked_seconds)  # time the embedder spent waiting on the store

Errors and shutdown

An exception in any process is reported to on_error, every other stage is cancelled, every stage_context is exited, and the original exception is raised from run(). External cancellation (a timeout around run(), Ctrl-C) exits the contexts the same way and is not reported as an error. A running sync process cannot be interrupted: the conveyor waits for it, so give blocking clients their own timeouts.

Invalid topologies and settings are rejected when the Conveyor is built, with a TopologyError that names the stage: fan-in, cycles, a second root, duplicate names, a batch stage without batch_size, a root with more than one worker, settings for a stage that is not in the topology.

Development

uv sync
uv run pytest
uv run ruff check .
uv run mypy

tests/test_e2e.py is a runnable demo of everything above.