Pattern 9: Batch Processing

The Problem: Per-Item vs Batch Overhead

Per-item: 1000 items = 1000 individual calls (network overhead, latency).

Batch: Accumulate 100 items, send 1 batch request. 1000 items = 10 batch calls.

Trade-off: Latency vs throughput. Batching increases per-item latency but increases total throughput.

Use case: Bulk inserts (database), batch API calls (GraphQL mutations), event aggregation.

Per-Item vs Batch Timeline

timeline title 1000 Items: Per-Item vs Batch Processing section Per-Item (10ms per call) Item 1: 10ms Item 2: 10ms Item 3-100: 990ms : ...continues... Item 1000: 10s total section Batch-100 (50ms per batch) Batch 1 (items 1-100): 50ms, wait to accumulate Batch 2 (items 101-200): 50ms, wait Batch 10 (items 901-1000): 50ms : 0.5s total (20× faster!)

The win: 20× faster throughput. Trade-off: individual items wait longer to batch (but batch processes faster overall).

Implementation: Batching Queue

sequenceDiagram participant User as User participant Batcher as Batcher participant Handler as Handler participant DB as Database User->>Batcher: add_item(id=1) Batcher->>Batcher: items=[1] User->>Batcher: add_item(id=2-10) Batcher->>Batcher: items=[1-10] Note over Batcher: 100ms passes or 10 items accumulated Batcher->>Handler: process_batch([1-10]) Handler->>DB: INSERT INTO users VALUES (...) DB-->>Handler: OK Handler->>Batcher: Done! Batcher->>Batcher: Clear items, restart User->>Batcher: add_item(id=11-20)

import asyncio
from typing import List, Callable

class Batcher:
    def __init__(self, batch_size: int, timeout: float, process_fn: Callable):
        self.batch_size = batch_size
        self.timeout = timeout
        self.process_fn = process_fn
        self.items = []
        self.lock = asyncio.Lock()

    async def add_item(self, item):
        """Add an item, trigger batch processing if needed."""
        async with self.lock:
            self.items.append(item)
            if len(self.items) >= self.batch_size:
                await self._flush()

    async def _flush(self):
        """Process accumulated items as a batch."""
        if not self.items:
            return
        batch = self.items[:]
        self.items = []
        await self.process_fn(batch)

    async def background_flusher(self):
        """Periodically flush if timeout expires."""
        while True:
            await asyncio.sleep(self.timeout)
            async with self.lock:
                await self._flush()

# Usage
async def process_batch(items: List[int]):
    print(f"Processing batch of {len(items)}: {items}")
    # Bulk insert to database
    await asyncio.sleep(0.1)  # Simulate DB operation

async def main():
    batcher = Batcher(batch_size=100, timeout=1.0, process_fn=process_batch)

    # Start background flusher
    flush_task = asyncio.create_task(batcher.background_flusher())

    # Send 250 items over time
    for i in range(250):
        await batcher.add_item(i)
        await asyncio.sleep(0.01)

    # Final flush
    async with batcher.lock:
        await batcher._flush()

    flush_task.cancel()

asyncio.run(main())

Real-World: FastAPI Batch Endpoint

from fastapi import FastAPI

app = FastAPI()
user_batcher = Batcher(batch_size=100, timeout=1.0, process_fn=insert_users_to_db)

@app.post("/users/batch")
async def create_user(user_data: dict):
    await user_batcher.add_item(user_data)
    return {"status": "queued"}

@app.on_event("shutdown")
async def on_shutdown():
    # Flush remaining items on shutdown
    async with user_batcher.lock:
        await user_batcher._flush()