Building Async Batch Processors for Serialization Events

High-throughput packaging lines emit serialized ObjectEvent, AggregationEvent, and TransactionEvent records faster than any single trading-partner endpoint will accept them, and synchronous request-response code stalls the whole line the moment a partner API throttles or times out. This page shows how to build a concrete async batch processor in Python that decouples event capture from transmission, chunks events into rate-limit-friendly batches, dispatches them with bounded concurrency, and routes exhausted retries to a dead-letter queue — all while preserving the deterministic, gap-free audit trail the Drug Supply Chain Security Act (DSCSA) demands. It is the implementation-level deep dive under the Async Batch Processing Pipelines reference architecture, which sits inside the broader Serialization Data Ingestion & EPCIS Event Sync domain.

Figure — Transmission state machine for async batch delivery. Terminal states carry a heavier border.

Prerequisites #

Before writing the processor, confirm the following are in place:

• Python 3.10+ — the code uses structural typing, match-friendly syntax, and modern asyncio idioms.
• Libraries: httpx (async HTTP with connection pooling and HTTP/2), pydantic v2 (payload validation with field_validator), and the standard-library asyncio, hashlib, and logging modules. aiohttp is a drop-in alternative to httpx if you already standardize on it.
• A populated SGTIN event source — commissioned units whose serial pool is active, so each event carries a real GTIN (01), serial (21), expiration date (17), and lot (10). Validating malformed EPCIS before it enters the pipeline is covered in Schema Validation & Error Handling; this page assumes events are already structurally sound.
• Trading-partner endpoint details: the partner’s EPCIS 2.0 capture URL, its documented rate limit (requests/second and burst allowance), and the destination GLN. If your target is a Verification Router Service rather than a capture interface, see the Verification Router Service Architecture for the request contract.
• A persistent dead-letter store — a database table, an object bucket, or a broker queue — so batches that exhaust their retries survive process restarts for later reconciliation.

Step-by-Step Solution #

Step 1 — Model the batch and derive a deterministic idempotency key #

Every batch needs a stable identifier that is a pure function of its contents. Hashing the sorted JSON payload (SHA-256, truncated) yields a key that is identical across retries and across process restarts, which lets the partner deduplicate re-sends. DSCSA rule satisfied: interoperable exchange must be gap-free and non-duplicative — a content-derived key gives you effectively-once delivery even when the network forces a resend after an unacknowledged 2xx.

import asyncio
import hashlib
import json
import logging
from dataclasses import dataclass
import httpx

logger = logging.getLogger("dscsa.batch_processor")

@dataclass
class EPCISBatch:
    batch_id: str
    events: list[dict]
    attempt: int = 0
    max_attempts: int = 5

    @staticmethod
    def make_id(events: list[dict]) -> str:
        payload = json.dumps(events, sort_keys=True).encode()
        return hashlib.sha256(payload).hexdigest()[:16]

Step 2 — Chunk the event stream into rate-limit-friendly batches #

Never send one event per request and never send the whole day’s production in a single call. Chunk by a size the partner’s rate limit and payload ceiling both tolerate. Group by destination first so each batch targets exactly one endpoint and jurisdiction. GS1 rule satisfied: each ObjectEvent in the eventList retains its own (01)/(21)/(17)/(10) identifiers, so chunking is purely a transport concern and never mutates event semantics.

def chunk_events(events: list[dict], size: int = 500) -> list[list[dict]]:
    """Split a validated event stream into fixed-size batches."""
    return [events[i : i + size] for i in range(0, len(events), size)]

Tune size against the partner’s documented limits: a 500-event chunk against a 5-request/second quota with five concurrent workers keeps you well inside a typical burst window while still draining a million-unit run in minutes rather than hours.

Step 3 — Dispatch a single batch with backoff and idempotent retry #

The core of the processor is a coroutine that sends one batch, interprets the response, and retries transient failures with exponential backoff. A 429 Too Many Requests is honored via its Retry-After header rather than the local backoff clock. DSCSA rule satisfied: a 429 or 5xx is treated as a first-class compliance signal, not a silent drop — the event is retried or dead-lettered, never lost. The same backoff discipline underpins handling rate limits on FDA verification APIs.

async def dispatch_batch(
    client: httpx.AsyncClient,
    batch: EPCISBatch,
    endpoint: str,
    sem: asyncio.Semaphore,
) -> bool:
    async with sem:
        backoff = 1.0
        while batch.attempt < batch.max_attempts:
            batch.attempt += 1
            try:
                resp = await client.post(
                    endpoint,
                    json={"epcisBody": {"eventList": batch.events}},
                    headers={"X-Idempotency-Key": batch.batch_id},
                    timeout=30.0,
                )
                if resp.status_code in (200, 202):
                    logger.info("Batch %s delivered on attempt %d",
                                batch.batch_id, batch.attempt)
                    return True
                if resp.status_code == 429:
                    retry_after = float(resp.headers.get("Retry-After", backoff))
                    await asyncio.sleep(retry_after)
                    backoff = min(backoff * 2, 60.0)
                    continue
                resp.raise_for_status()
            except httpx.RequestError as exc:
                logger.warning("Batch %s request error: %s", batch.batch_id, exc)
                await asyncio.sleep(backoff)
                backoff = min(backoff * 2, 60.0)

        logger.error("Batch %s exhausted retries — routing to DLQ", batch.batch_id)
        return False

Step 4 — Bound concurrency and fan out with a semaphore #

An unbounded asyncio.gather over thousands of batches will open thousands of sockets and instantly trip the partner’s rate limit. An asyncio.Semaphore caps in-flight requests, applying backpressure so the pipeline self-throttles to the partner’s quota. Operational rule satisfied: controlled concurrency keeps the packaging line decoupled — a slow partner drains the queue slower but never blocks event capture.

async def process_batches(
    event_chunks: list[list[dict]],
    endpoint: str,
    max_concurrent: int = 5,
) -> list[tuple[EPCISBatch, bool]]:
    sem = asyncio.Semaphore(max_concurrent)
    batches = [
        EPCISBatch(batch_id=EPCISBatch.make_id(chunk), events=chunk)
        for chunk in event_chunks
    ]
    async with httpx.AsyncClient(http2=True) as client:
        results = await asyncio.gather(
            *[dispatch_batch(client, b, endpoint, sem) for b in batches]
        )
    return list(zip(batches, results))

Step 5 — Persist failures to a dead-letter queue and record terminal state #

Batches that return False have exhausted their retries and must be captured for reconciliation, never discarded. Persist the full payload plus its batch_id and terminal state so an operator or an automated drain job can replay it. DSCSA rule satisfied: the six-year retention and gap-free traceability obligations require that every event reach a terminal ACKNOWLEDGED or FAILED state that is durably recorded.

async def run_pipeline(events: list[dict], endpoint: str) -> None:
    chunks = chunk_events(events, size=500)
    outcomes = await process_batches(chunks, endpoint, max_concurrent=5)
    for batch, ok in outcomes:
        state = "ACKNOWLEDGED" if ok else "FAILED"
        logger.info("Batch %s final state: %s", batch.batch_id, state)
        if not ok:
            await dead_letter_store.put(batch.batch_id, batch.events)  # your DLQ

if __name__ == "__main__":
    asyncio.run(run_pipeline(load_commissioned_events(), TRADING_PARTNER_URL))

Verification #

Confirm the processor behaves correctly before pointing it at a live partner:

• Unit-test the retry path with a mock transport. Assert that a 429 followed by a 200 results in exactly two attempts and returns True, and that five consecutive 503s return False and enqueue to the DLQ.

import pytest, httpx

@pytest.mark.asyncio
async def test_retry_then_success():
    responses = iter([httpx.Response(429, headers={"Retry-After": "0"}),
                      httpx.Response(202)])
    transport = httpx.MockTransport(lambda req: next(responses))
    async with httpx.AsyncClient(transport=transport) as client:
        batch = EPCISBatch(batch_id="abc123", events=[{"eventID": "1"}])
        ok = await dispatch_batch(client, batch, "https://partner/capture",
                                  asyncio.Semaphore(1))
    assert ok and batch.attempt == 2

• Assert idempotency-key stability: hash the same event list twice and confirm the batch_id is identical; reorder the events and confirm it is still identical (because the payload is sorted before hashing).
• Inspect the audit log after a dry run: every batch_id should appear with exactly one terminal ACKNOWLEDGED or FAILED line, and the count of delivered events plus dead-lettered events must equal the input count — the reconciliation invariant that proves no event was silently dropped.

Gotchas & Edge Cases #

• Idempotency-key collisions from non-canonical JSON. If any producer emits floats, differently-ordered keys, or unicode escapes, json.dumps(..., sort_keys=True) may still differ. Normalize types upstream (or use a canonical JSON serializer) so semantically identical batches always hash the same.
• UTC vs. local time in eventTime. EPCIS requires eventTime with an explicit eventTimeZoneOffset. A batch whose events silently carry local time will pass transport but fail partner validation and land in the DLQ — validate timezone offsets before chunking, not after transmission.
• Retry-After as an HTTP-date, not seconds. Some gateways return Retry-After as an RFC 7231 date string; float(...) will raise. Parse both forms, and cap any absurd value so a misconfigured partner can’t stall the pipeline for hours.
• Unacknowledged 2xx after a dropped connection. If the partner committed the batch but the ack never reached you, a naive retry double-sends. The stable X-Idempotency-Key is what lets the partner deduplicate — never regenerate it on retry.
• Semaphore released but socket exhausted. A max_concurrent far above the partner’s real quota just moves the bottleneck into their 429 handling. Size the semaphore to the documented rate limit, and prefer HTTP/2 connection reuse over opening fresh sockets per batch.

FAQ #

Should I use a persistent broker instead of an in-memory queue? #

For a single-run export the in-memory chunk list shown here is sufficient. For continuous, always-on ingestion — where the process may restart mid-drain — back the pipeline with a durable broker (Kafka or RabbitMQ) so unsent batches survive a crash. That streaming variant belongs to Real-Time Event Stream Processing rather than the batch pattern on this page.

How large should each batch be? #

Start at 500 events and adjust to the partner’s documented payload ceiling and rate limit. Larger batches reduce per-request overhead but increase the blast radius of a single rejected call; smaller batches isolate failures at the cost of more round trips. Measure the partner’s 429 frequency and tune size and max_concurrent together.

Where does schema validation belong — before or inside the processor? #

Before. Validate structure and identifier completeness at ingestion so the transmission stage only ever sees well-formed events; a batch that fails partner validation after transmission is far more expensive to reconcile. See Schema Validation & Error Handling for the Pydantic and lxml patterns.

Related #

• Parent guide: Async Batch Processing Pipelines
• Handling Rate Limits on FDA Verification APIs
• Parsing EPCIS XML with Python lxml Efficiently
• Real-Time Event Stream Processing
• Pillar: Serialization Data Ingestion & EPCIS Event Sync