AsyncBatcher

Documentation · Release notes

AsyncBatcher combines concurrent Python requests into batches and routes each result back to its caller. Wrap an existing bulk function, subclass AsyncBatcher, or use an adapter for a supported database, messaging client, or prediction model. The core package has no runtime dependencies.

Start with your workload

Start with a bounded importer or interactive service. The workload guide helps choose limits, preserve routing metadata, and decide whether batching fits your backend. Batching requires a useful bulk operation; it is not a rate limiter, durable queue, or exactly-once delivery system.

Install

Python 3.11 or newer is required. This README targets Async Batcher 0.3.0; check the release history for publication status. Install the package into your application's environment:

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

The core has no runtime dependencies. Add only the extras you need, for example:

python -m pip install 'async-batcher[aws]==0.3.0'

See installation for uv, other extras, and repository recipes, and migration from 0.2.2 before upgrading. Core and adapter usage does not need a source checkout. Runnable files under examples/ are repository recipes, with separate setup instructions.

Quickstart

Save this as demo.py and run python demo.py in the environment where you installed the package:

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())

process() is a coroutine: submit calls concurrently to form batches. A loop that awaits each call before submitting the next normally produces single-item batches. Processing starts automatically. Stop the batcher before closing its clients, model resources, or executor. Existing subclasses and adapters support the same managed lifecycle.

Guides

• Compatibility and migration: published features, Python requirements, and upgrade checks.
• Bounded starting configurations: copyable importer and interactive service examples.
• Choose by workload: request compatibility, safe metadata, and configuration ownership.
• Batch an existing function: callable processors and managed cleanup.
• Bulk and streaming producers: bounded maps and waiting for queue space.
• Cost budgets: byte, token, or work limits alongside batch counts.
• Variable-length inference: padding, conservative token budgets, and per-request output shapes.
• Operational visibility: snapshots and payload-free completion events.
• Grouped batching: isolate compatible requests with bounded child ownership.
• In-flight coalescing: share equivalent concurrent requests with bounded ownership.
• Cookbook: runnable SQLite imports, embedding pipelines, and measured batch sizes.
• Resumable imports: immutable JSONL input with atomic SQLite checkpoints.
• Rejected-record imports: continue past known input errors and retain a metadata-only audit.
• Durable embedding handoff: save vectors before delivery and replay unfinished upserts.
• Provider quota and retry policy: shared pacing, finite throttling retries, and cancellation.
• Cohere to Pinecone: real embeddings into an existing compatible vector index.
• FastAPI recipe: lifespan-managed predictions, SQLite writes, and HTTP overload handling.
• Operational recipes: separate deadlines, shared backend capacity, thread lifetime, and bounded telemetry.
• Core API and lifecycle: limits, result mapping, errors, shutdown, restart, and executors.
• Contributing: locked development environments, test selection, integration services, and CI.
• Benchmarks: reproducible core measurements and archived application results.
• Public type contracts: inferred results, typed options, and installed-package checks.
• Security reporting and Code of Conduct.

Optional adapters

The eight extras are independent; importing async_batcher does not import their clients. All adapters use the core lifecycle, with the limits and resource ownership described in each guide.

Services that do not offer a suitable bulk operation need a custom process_batch implementation. Batching alone does not enforce a service's request-rate limit or guarantee exactly-once side effects.