Metadata-Version: 2.3
Name: async-conveyor
Version: 0.3.0
Summary: Run a tree of sync/async/batch stages concurrently on anyio, with bounded queues between them
Requires-Dist: anyio>=4.15.1
Requires-Python: >=3.11
Description-Content-Type: text/markdown

# async-conveyor

Run a tree of processing stages concurrently on [anyio](https://anyio.readthedocs.io/), 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.

```python
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.

```sh
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:

|                   | one item per call | a batch of items per call |
|-------------------|-------------------|---------------------------|
| event loop (`async def`) | `AsyncStage`      | `AsyncBatchStage`         |
| thread (`def`)           | `SyncStage`       | `SyncBatchStage`          |

`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.

```python
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.

```python
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`.

```python
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.

| setting         | default | meaning                                                                                   |
|-----------------|---------|-------------------------------------------------------------------------------------------|
| `workers`       | 1       | items processed concurrently: coroutines for async stages, threads for sync ones. Root: 1 |
| `buffer_size`   | 1       | the stage blocks when a queue into one of its children holds this many items              |
| `batch_size`    | –       | batch stages only, required: `process` gets up to this many items                         |
| `batch_timeout` | `None`  | batch stages only: seconds to wait for a full batch after its first item. `None`: forever |

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.

```python
@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:

```python
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:

```python
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

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

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