Async Batcher

Combine concurrent Python requests into bounded batches and route each result back to its caller. Wrap an existing bulk function, subclass AsyncBatcher, or use an adapter for database writes, messaging, and model inference. The core has no runtime dependencies and requires Python 3.11+.

Start with your workload

You need to…	Start here
Batch predictions or writes in a web service	Bounded service configuration, then FastAPI
Import a file or consume an async stream	Bounded maps and import/recovery choices
Use an existing database or messaging client	Database adapters, DynamoDB, AWS streams, or Kafka
Send embeddings to a vector store	Embedding pipeline and durable handoff
Separate tenants, model revisions, or destinations	Compatible request groups
Limit bytes, tokens, or estimated work	Cost budgets and variable-length inference

Batching needs a useful bulk operation. It does not add durable delivery or exactly-once writes. The workload guide explains when batching helps and how to choose limits.

Install

These examples target Async Batcher 0.3.0 and require Python 3.11 or newer. The shared release notes show which versions are published. Install into your application's environment:

python -m pip install 'async-batcher==0.3.0'

The core needs no optional clients. Use installation to select extras or add the package with uv, and read migration from 0.2.2 when upgrading. Repository recipes have their own setup.

Your first batch

Save this as demo.py and run python demo.py in your package environment:

import asyncio

from async_batcher import AsyncBatcher


async def double_many(values: list[int]) -> list[int]:
    return [value * 2 for value in values]


async def main():
    async with AsyncBatcher.from_callable(
        double_many, max_batch_size=5, max_queue_size=20, max_queue_time=0.01, concurrency=2
    ) as batcher:
        results = await asyncio.gather(*(batcher.process(i) for i in range(10)))
        print(results)  # [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]


asyncio.run(main())

Submit process() calls concurrently to form batches. Awaiting each request before submitting the next normally produces single-item batches. For large inputs, use bounded maps instead of creating a task for every item. The async context manager drains accepted work before exiting; keep clients and executors alive until that cleanup finishes.

Continue with functions and cleanup, the two bounded configurations, or the API and lifecycle reference.