Python Asyncio Patterns for Batch DVIR Processing

A synchronous Driver Vehicle Inspection Report (DVIR) pipeline answers how to persist one inspection; it does not answer how to persist five thousand of them inside the fifteen-minute end-of-shift window without dropping a federal record, and that gap is where carriers get burned. 49 CFR § 396.11(a)(2) requires the motor carrier to receive and retain a report for every driver-operated vehicle, and a submission that silently times out behind a serialized database connection is not a lost log line — it is a missing federal record and a potential out-of-service (OOS) exposure at the next roadside inspection. This page answers one focused question: which concrete asyncio patterns turn a validated batch of DVIR records into bounded-concurrency, partial-failure-safe writes without exhausting a connection pool or silently swallowing a compliance error. It is the runnable implementation of the concurrency layer specified by the parent Async Batching for High-Volume Ingestion reference, and it sits inside the broader DVIR Ingestion & Digital/Paper Parsing Workflows architecture.

Get the concurrency bound wrong and you get one of two silent failures: unbounded coroutines that exhaust the PostgreSQL connection pool and stall every worker, or an unhandled exception in one record that cancels the whole batch and drops thousands of valid inspections. Both are audit failures. The patterns below make concurrency explicit, make partial failure the default, and keep every rejected record on a reconstructable path.

This layer is a pure concurrency orchestrator: it receives records that already conform to the canonical DVIR contract and returns a per-record outcome. It does not parse OCR, it does not normalize fields, and it does not score defects. Before implementing it you need Python 3.11 or later (every code block uses 3.11+ syntax — asyncio.TaskGroup, ExceptionGroup, and except* all require it) and:

• asyncpg>=0.29 — pooled, non-blocking PostgreSQL access for the append-only audit ledger; the pool size is the hard ceiling every concurrency calculation is derived from.
• pydantic>=2.6 — strict validation against the canonical record shape defined by the Standardized DVIR JSON Schema Design; a record that fails that contract is rejected here, never defaulted through.
• redis>=5.0 (via redis.asyncio) — durable dead-letter queue and idempotency-key store so a quarantined record survives a worker restart.
• structlog>=24.1 — correlation-ID-aware structured logging; interleaved print() output is unreadable under concurrency and useless for audit reconstruction.

Records reach this layer already normalized: paper submissions are pre-processed by the PDF & Image OCR Pipeline Setup, digital exports enter through the Mobile App DVIR Export Integration endpoint, and inconsistent driver-entered fields are already reconciled by Normalizing Inconsistent Driver Input Fields. Defect codes arrive normalized against Defect Code Standardization Across Fleets.

The concurrency bound is not a taste decision; it is arithmetic derived from your PostgreSQL pool. If each record consumes queries_per_record connections for the duration of its coroutine, then firing more than pool_size / queries_per_record records concurrently guarantees a TooManyConnectionsError at the shift-change peak. Compute the ceiling once, apply a 20% safety margin for pool churn, and never hard-code the number.

import math


def max_concurrency(pool_size: int, queries_per_record: int) -> int:
    """Concurrency ceiling from the DB pool, with a 20% safety margin.

    pool_size            -- asyncpg pool max_size (the hard ceiling)
    queries_per_record   -- connections a single record holds concurrently
    """
    if queries_per_record < 1:
        raise ValueError("queries_per_record must be >= 1")
    raw = pool_size / queries_per_record
    # 20% headroom absorbs pool churn during the end-of-shift submission spike.
    return max(1, math.floor(raw * 0.80))

Under 49 CFR § 396.11 every record must resolve to a known terminal state: committed to the audit ledger, or quarantined with a reason. A coroutine that raises past the batch boundary erases that guarantee, so wrap every record in an explicit, immutable outcome. Carry the correlation_id and idempotency_key so the outcome is traceable and duplicate-safe.

from dataclasses import dataclass
from enum import StrEnum


class Outcome(StrEnum):
    COMMITTED = "committed"      # persisted to the append-only ledger
    QUARANTINED = "quarantined"  # routed to the dead-letter queue, § 396.11 record preserved


@dataclass(frozen=True, slots=True)
class RecordResult:
    correlation_id: str
    idempotency_key: str
    outcome: Outcome
    reason: str | None = None   # populated only when QUARANTINED

The asyncio.Semaphore must be acquired inside the per-record coroutine, not around the whole batch — acquiring it per record is what actually bounds concurrent DB access. Trap the exception at the record boundary so one malformed VIN, one missing certification timestamp, or one corrupted attachment quarantines a single record instead of taking down the batch. Idempotency is checked first: a re-synced offline submission must not double-write.

import asyncpg
import structlog
from redis.asyncio import Redis

log = structlog.get_logger()


async def process_record(
    record: dict,
    sem: asyncio.Semaphore,
    pool: asyncpg.Pool,
    redis: Redis,
) -> RecordResult:
    """Validate + persist ONE record under the concurrency bound.

    Never raises past this boundary: every path returns a RecordResult so the
    batch preserves a terminal state for every § 396.11 record.
    """
    cid = record["correlation_id"]
    key = record["idempotency_key"]
    async with sem:  # bound concurrent DB access to max_concurrency
        try:
            # Duplicate suppression: a re-synced offline DVIR must not double-write.
            if await redis.set(f"idem:{key}", cid, nx=True, ex=86_400) is None:
                return RecordResult(cid, key, Outcome.COMMITTED, reason="duplicate-suppressed")

            dvir = DvirRecord.model_validate(record)  # Pydantic; raises on bad payload
            async with pool.acquire() as conn:
                await conn.execute(
                    "INSERT INTO dvir_ledger (correlation_id, vin, driver_id, "
                    "received_at, payload) VALUES ($1, $2, $3, $4, $5)",
                    cid, dvir.vin, dvir.driver_id, dvir.received_at, record,
                )
            return RecordResult(cid, key, Outcome.COMMITTED)

        except Exception as exc:  # quarantine, never drop — § 396.11 record must survive
            await redis.rpush("dvir:dlq", json.dumps({"cid": cid, "record": record,
                                                       "reason": repr(exc)}))
            await redis.delete(f"idem:{key}")  # release key so a fixed retry can re-enter
            log.warning("dvir.quarantined", correlation_id=cid, reason=repr(exc))
            return RecordResult(cid, key, Outcome.QUARANTINED, reason=repr(exc))

For ingestion you want partial success: a bad record must not cancel its neighbors. Use asyncio.gather(..., return_exceptions=True) so any exception that escapes Step 3 is returned as a value rather than propagated. Because Step 3 already traps and converts exceptions to RecordResult, return_exceptions=True here is a belt-and-suspenders guard against a bug in the worker itself.

import json


async def process_batch(
    batch: list[dict],
    pool: asyncpg.Pool,
    redis: Redis,
    queries_per_record: int = 1,
) -> list[RecordResult]:
    """Fan out one batch under a semaphore; partial success by design."""
    sem = asyncio.Semaphore(max_concurrency(pool.get_max_size(), queries_per_record))
    results = await asyncio.gather(
        *(process_record(r, sem, pool, redis) for r in batch),
        return_exceptions=True,  # a worker-level bug becomes a value, not a batch abort
    )

    final: list[RecordResult] = []
    for r, raw in zip(batch, results):
        if isinstance(raw, RecordResult):
            final.append(raw)
        else:  # a RecordResult was never produced — treat as quarantined, don't drop
            await redis.rpush("dvir:dlq", json.dumps(
                {"cid": r["correlation_id"], "record": r, "reason": repr(raw)}))
            final.append(RecordResult(r["correlation_id"], r["idempotency_key"],
                                      Outcome.QUARANTINED, reason=repr(raw)))
    return final

asyncio.TaskGroup (Python 3.11+) gives structured cancellation: if one child raises, all siblings are cancelled and the errors surface as an ExceptionGroup. That is the wrong default for ingestion — you do not want a single bad VIN cancelling four thousand valid inspections. Reserve TaskGroup for genuinely atomic units of work, such as writing a record and its signature-verification receipt together, where a partial write would corrupt the audit chain. Handle the group with except*.

async def commit_atomic(record: dict, receipt: dict, pool: asyncpg.Pool) -> None:
    """All-or-nothing: the ledger row and its signature receipt commit together."""
    async with pool.acquire() as conn, conn.transaction():
        try:
            async with asyncio.TaskGroup() as tg:
                tg.create_task(_write_ledger(conn, record))
                tg.create_task(_write_receipt(conn, receipt))
        except* asyncpg.PostgresError as eg:
            # Any child failure rolls back the transaction; re-raise for the caller.
            raise RuntimeError(f"atomic commit failed: {eg.exceptions}") from eg

OCR callouts and signature-verification APIs fail transiently. Retry them, but cap at three attempts: an unbounded retry loop can stall a record past the point where its § 396.11 certification window matters, turning a transient blip into an audit gap. Retry only transient errors — never retry a ValidationError, because a malformed payload will fail identically forever and belongs in the dead-letter queue immediately.

async def with_backoff(coro_factory, *, attempts: int = 3, base: float = 0.5):
    """Retry a coroutine factory on transient errors; cap attempts to avoid audit gaps."""
    for attempt in range(1, attempts + 1):
        try:
            return await coro_factory()
        except (asyncpg.PostgresConnectionError, asyncio.TimeoutError) as exc:
            if attempt == attempts:
                raise  # exhausted -> caller quarantines, § 396.11 record preserved
            await asyncio.sleep(base * 2 ** (attempt - 1))  # 0.5s, 1.0s, ...

The batch is a compliance artifact, so test the invariants that keep records off the floor, not merely that the code runs. Three assertions matter most: a poison record never cancels its batch, the semaphore genuinely bounds concurrency, and a duplicate never double-writes.

import pytest


@pytest.mark.asyncio
async def test_poison_record_does_not_cancel_batch(pool, redis):
    # One record has a bad VIN; the other two are valid and must still commit.
    batch = [_valid(), _bad_vin(), _valid()]
    results = await process_batch(batch, pool, redis)
    outcomes = [r.outcome for r in results]
    assert outcomes.count(Outcome.COMMITTED) == 2       # neighbors survive
    assert outcomes.count(Outcome.QUARANTINED) == 1     # poison isolated
    assert await redis.llen("dvir:dlq") == 1            # § 396.11 record preserved


@pytest.mark.asyncio
async def test_semaphore_bounds_concurrency(pool, redis, monkeypatch):
    # Instrument the worker to record peak in-flight count.
    peak = {"n": 0, "cur": 0}

    async def _tracked(*a, **k):
        peak["cur"] += 1
        peak["n"] = max(peak["n"], peak["cur"])
        await asyncio.sleep(0.01)
        peak["cur"] -= 1
        return RecordResult("c", "k", Outcome.COMMITTED)

    monkeypatch.setattr("mymod.process_record", _tracked)
    await process_batch([_valid() for _ in range(100)], pool, redis, queries_per_record=1)
    assert peak["n"] <= max_concurrency(pool.get_max_size(), 1)  # bound never exceeded


@pytest.mark.asyncio
async def test_idempotency_key_suppresses_duplicate(pool, redis):
    rec = _valid()
    first = await process_batch([rec], pool, redis)
    second = await process_batch([rec], pool, redis)  # same idempotency_key
    assert first[0].outcome is Outcome.COMMITTED
    assert second[0].reason == "duplicate-suppressed"  # no second ledger row

Run these under pytest-asyncio in CI and fail the build on any regression. The poison-record test is the one that catches the most dangerous class of bug: a fast path that quietly aborts the batch on the first exception and drops federal records.

• Semaphore acquired around the batch, not the record. Wrapping async with sem: around the whole gather call bounds nothing — every coroutine still opens a connection at once and exhausts the pool. The semaphore must be acquired inside process_record, once per record, so the count reflects concurrent DB access. This is the single most common asyncio DVIR bug.
• gather without return_exceptions=True. The default asyncio.gather propagates the first exception and leaves the remaining coroutines running but orphaned — their results are lost and their records silently dropped. For ingestion, always pass return_exceptions=True and inspect each result, or a single bad VIN erases every valid inspection submitted in the same shift.
• Retrying non-transient errors. A ValidationError on a malformed payload will fail identically on every retry, so a blanket retry wrapper burns the backoff budget and delays quarantine. Retry only connection and timeout errors; route validation failures straight to the dead-letter queue on the first attempt.
• Offline re-sync double-writing. A mobile app that re-sends its queue on reconnect submits the same DVIR twice; without an idempotency key checked before the write, both rows land in the ledger and inflate downstream recurrence counts. Set the key with nx=True and reconcile the record against the inspection timestamp, not the sync time. Release the key on quarantine so a corrected retry can re-enter.

• Async Batching for High-Volume Ingestion — the parent reference specifying the batching layer this page implements the concurrency for.
• PDF & Image OCR Pipeline Setup — the upstream stage that feeds paper-sourced records into these batches.
• Best Practices for Mobile DVIR API Integration — the digital export endpoint whose offline re-sync drives the idempotency requirement.
• Standardized DVIR JSON Schema Design — the canonical record shape every batched record is validated against.
• Building a Weighted Defect Scoring Model in Python — the downstream consumer that scores the committed records this layer persists.

Part of the Async Batching for High-Volume Ingestion guide. Back to DVIR Ingestion & Digital/Paper Parsing Workflows.