Implementing Async Batch Processing for High-Volume Permit Submissions

17 min read 2 sub-topics

Asynchronous batch processing is the throughput backbone of the broader Automated Permit Ingestion and Parsing Workflows pipeline: it decouples the moment a submission is received from the slower work of enrichment, validation, and persistence so the public-facing layer stays responsive even during seasonal surges. This page walks Python automation builders through a production-grade async batch pipeline — chunking submissions into idempotent units, publishing them to a durable broker, consuming them concurrently with backpressure, enriching and validating each record, then committing with deterministic state transitions and routing failures to a quarantine queue.

Municipal permitting offices face predictable but intense submission surges driven by seasonal construction windows, zoning ordinance revisions, and state-mandated reporting deadlines. Synchronous ingestion pipelines degrade rapidly under concurrent load, producing HTTP 504 gateway timeouts, orphaned database transactions, and fragmented audit trails that complicate compliance reviews. Moving to an asynchronous batch architecture turns those unpredictable spikes into a controlled, auditable, and horizontally scalable workflow.

Async batch processing pipeline for high-volume permit submissions Submissions arrive from a public portal, nightly SFTP drops, and inter-agency exchanges. A chunker aggregates them into fixed-size batches keyed by a sha256 idempotency digest and writes raw payloads to an S3 object store. Batch envelopes are published to a durable broker (RabbitMQ or SQS) across critical, standard, and maintenance priority lanes. An asyncio worker pool, bounded by a Semaphore of 32 and broker prefetch of 4, consumes each batch and runs enrich, validate, then commit per record. Committed records land in PostgreSQL under a unique idempotency-key constraint as accepted; permanent validation failures route to a dead-letter quarantine queue. Every state transition is written to an append-only audit log. INGEST · CHUNK · PUBLISH CONSUME · ENRICH · VALIDATE · COMMIT Public portal JSON burst SFTP drop nightly files Inter-agency bulk reconcile Chunker fixed-size · sha256 key raw payload → S3 Durable broker RabbitMQ / SQS · persistent critical standard maintenance envelope asyncio worker pool Semaphore(32) · QoS prefetch 4 — bounded backpressure Enrich parcel · zoning · fees asyncio.gather Validate pydantic schema Commit atomic txn ON CONFLICT consume PostgreSQL unique(idem_key) Dead-letter queue permanent fault → quarantine accepted schema fail · no retry Append-only audit log · permit_audit received → chunked → enriched → validated → accepted · rejected · quarantined
End-to-end async batch pipeline: portal, SFTP, and inter-agency submissions are chunked into idempotent batches, published to a durable broker, consumed by a bounded asyncio worker pool, and committed exactly-once — with permanent failures quarantined and every transition audited.

Problem Statement and Scope

A permit submission rarely arrives one record at a time. Contractors drop nightly SFTP files of several hundred applications, public portals burst at the start of a fee cycle, and inter-agency exchanges push bulk reconciliations. A synchronous request/response handler tries to receive, enrich, validate, and persist each record inside one HTTP transaction. Under load this design fails in three recurring ways:

  • Gateway timeouts. A batch that takes 90 seconds to enrich against a slow county GIS service blows past the load balancer’s 30-second idle limit; the client retries, and the same payload is processed twice.
  • Orphaned transactions. A worker crash mid-batch leaves rows half-written with no record of which applications committed, forcing manual reconciliation against the source system.
  • Backpressure collapse. With no bound on concurrency, a surge spawns unbounded coroutines and database connections until the worker is OOM-killed or the connection pool is exhausted.

Who is affected. Python builders own pipeline reliability and on-call recovery; municipal clerks and front-desk staff feel the latency when the same workers serve interactive lookups; compliance officers need a verifiable audit trail proving every submission was processed exactly once or explicitly quarantined.

Inputs and outputs. The pipeline consumes raw submission payloads (JSON from portals, delimited files from SFTP, scanned-document references) and produces committed permit records with a deterministic terminal state — accepted, rejected, or quarantined — plus structured run logs capturing batch identifiers, idempotency keys, per-stage latency, and final disposition for every record.

Prerequisites

This pipeline targets Python 3.10+ (the examples use match statements, X | Y type unions, and asyncio.TaskGroup-style patterns). A single broker and database node are enough for development; production runs a replicated broker and a connection-pooled database.

# Core async stack
pip install "aio-pika>=9.4"        # async AMQP client for RabbitMQ
pip install "asyncpg>=0.29"        # async PostgreSQL driver with native pooling
pip install "redis>=5.0"           # async client for idempotency keys
pip install "pydantic>=2.6"        # schema validation for submission payloads
pip install "tenacity>=8.2"        # retry policies for transient broker/DB faults

Environment assumptions: a durable message broker (RabbitMQ, Redis Streams, or AWS SQS) reachable from the worker subnet; an object store (S3, MinIO, or Azure Blob) for raw payloads too large to inline; a PostgreSQL database with a unique constraint on the idempotency key; and read access to whatever municipal registries the enrichment stage queries (parcel, zoning, fee schedule). Workers should run on their own pool, scaled independently of the public API gateway, so daytime counter traffic is never starved by an overnight batch.

Stage 1 — Chunk Submissions Into Idempotent Batches

The first design principle is idempotent chunking. Incoming payloads are aggregated into discrete batches bounded by a configurable threshold — record count, payload size, or a temporal window — and each batch carries an idempotency key derived from a deterministic digest of its contents. That key is what prevents duplicate processing when a retry, load-balancer timeout, or upstream failure redelivers the same data.

from __future__ import annotations

import hashlib
import json
from collections.abc import Iterator
from dataclasses import dataclass, field

MAX_RECORDS = 500          # batch boundary by record count
MAX_BYTES = 25 * 1024 * 1024  # 25 MB payload ceiling per batch


@dataclass(slots=True)
class Batch:
    records: list[dict] = field(default_factory=list)
    nbytes: int = 0

    @property
    def idempotency_key(self) -> str:
        # Deterministic digest of the ordered payload — identical content
        # always yields the same key, so a redelivered batch is a no-op.
        canonical = json.dumps(self.records, sort_keys=True, separators=(",", ":"))
        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def chunk(submissions: Iterator[dict]) -> Iterator[Batch]:
    """Yield fixed-boundary batches without loading the whole stream into RAM."""
    current = Batch()
    for record in submissions:
        size = len(json.dumps(record, separators=(",", ":")).encode("utf-8"))
        # Flush before the record would breach either boundary.
        if current.records and (
            len(current.records) >= MAX_RECORDS or current.nbytes + size > MAX_BYTES
        ):
            yield current
            current = Batch()
        current.records.append(record)
        current.nbytes += size
    if current.records:
        yield current

Streaming the source with a generator keeps memory flat regardless of file size — critical on the constrained virtual machines many municipalities run. Where chunk boundaries must respect data-residency rules, partition first by jurisdiction_fips so a single batch never mixes counties.

Stage 2 — Publish Batches to a Durable Broker

Each batch is wrapped in a lightweight routing envelope: metadata, processing directives, and a pointer to the raw payload in the object store rather than the payload itself. Publishing the envelope (not the bytes) keeps broker messages small and lets workers stream large payloads on demand. The publish must be durable — persistent delivery mode plus publisher confirms — so an accepted submission is never lost to a broker restart.

import aio_pika
from pydantic import BaseModel


class BatchEnvelope(BaseModel):
    batch_id: str
    idempotency_key: str
    jurisdiction_fips: str
    record_count: int
    payload_uri: str          # s3://permits-raw/2026/06/<batch_id>.json
    priority: str = "standard"  # critical | standard | maintenance


async def publish(channel: aio_pika.abc.AbstractChannel, env: BatchEnvelope) -> None:
    message = aio_pika.Message(
        body=env.model_dump_json().encode("utf-8"),
        delivery_mode=aio_pika.DeliveryMode.PERSISTENT,  # survives broker restart
        message_id=env.batch_id,
        headers={"idempotency_key": env.idempotency_key},
    )
    # Route by priority so overnight maintenance work never blocks live intake.
    await channel.default_exchange.publish(
        message, routing_key=f"permit.ingest.{env.priority}"
    )

Declaring the consuming queue with x-dead-letter-exchange and x-message-ttl arguments ensures malformed or expired envelopes are routed to quarantine instead of silently blocking the queue head. For overnight-specific queue topology, worker lifecycles, and broker tuning, see Managing Celery Task Queues for Overnight Batch Imports.

Stage 3 — Consume Concurrently With Bounded Backpressure

The consumer is where async concurrency earns its keep. Instead of blocking on each record, the worker fans records out across a bounded set of coroutines. The bound is the whole point: an asyncio.Semaphore caps in-flight work so a surge cannot exhaust memory or the database connection pool. Setting the broker prefetch (QoS) limit applies the same backpressure at the message layer.

import asyncio

MAX_CONCURRENCY = 32  # cap in-flight records per worker process


async def consume(
    queue: aio_pika.abc.AbstractQueue,
    pool,                       # asyncpg pool, shared across coroutines
    process_record,             # coroutine: (record, conn) -> TerminalState
) -> None:
    sem = asyncio.Semaphore(MAX_CONCURRENCY)

    async def handle(record: dict) -> None:
        async with sem:                      # backpressure: never exceed the cap
            async with pool.acquire() as conn:
                await process_record(record, conn)

    async with queue.iterator() as messages:
        async for message in messages:
            async with message.process(requeue=True):  # ack on success, requeue on crash
                env = BatchEnvelope.model_validate_json(message.body)
                records = await load_payload(env.payload_uri)  # stream from object store
                # Run the batch's records concurrently, bounded by the semaphore.
                await asyncio.gather(*(handle(r) for r in records))

Set the channel QoS so the broker hands a worker no more than a few batches at once (await channel.set_qos(prefetch_count=4)); combined with the semaphore this gives two independent throttles. Because worker pools scale horizontally, throughput grows by adding processes — not by raising the per-process concurrency past what the database can absorb.

Stage 4 — Enrich and Validate Each Record

Each record is enriched against external registries — parcel data, zoning classifications, fee schedules — then validated against the jurisdiction’s schema before it is allowed to commit. Enrichment lookups run concurrently and cache aggressively to stay inside municipal API rate limits; where a warmed read layer already holds that context, the lookup collapses to a cache hit (see Cache Warming Strategies for Permit Lookup APIs).

from pydantic import BaseModel, ValidationError


class PermitSubmission(BaseModel):
    submission_id: str
    permit_type_code: str
    parcel_id: str
    jurisdiction_fips: str
    project_valuation: float


async def enrich_and_validate(record: dict, registries) -> PermitSubmission:
    # Fire independent lookups concurrently rather than serially.
    zoning, fees = await asyncio.gather(
        registries.zoning_for(record["parcel_id"]),
        registries.fee_schedule(record["permit_type_code"]),
    )
    record = record | {"zoning_code": zoning, "fee_cents": fees}
    # Schema validation raises on missing statutory fields — caller routes to DLQ.
    return PermitSubmission.model_validate(record)

Validation rules are jurisdiction-specific: a parcel that fails zoning cross-reference, a fee code with no active schedule entry, or a missing statutory field is a permanent error and must not be retried. Distinguishing those from transient faults is the core of Error Handling and Retry Logic for Ingestion Pipelines, and the same triage governs how this pipeline routes failures in Stage 5.

Stage 5 — Commit Atomically and Record State Transitions

The terminal stage writes the record inside an explicit transaction guarded by the idempotency key, then logs the state transition. The unique constraint on the key is what makes the whole pipeline exactly-once: a redelivered record hits the constraint, the insert is a no-op, and no duplicate fee is assessed.

from enum import StrEnum


class TerminalState(StrEnum):
    ACCEPTED = "accepted"
    REJECTED = "rejected"
    QUARANTINED = "quarantined"


async def commit(record: dict, conn) -> TerminalState:
    async with conn.transaction():  # atomic: all-or-nothing per record
        # ON CONFLICT DO NOTHING makes a redelivered record idempotent.
        inserted = await conn.fetchval(
            """
            INSERT INTO permits (idempotency_key, submission_id, payload, state)
            VALUES ($1, $2, $3, 'accepted')
            ON CONFLICT (idempotency_key) DO NOTHING
            RETURNING id
            """,
            record["idempotency_key"], record["submission_id"], json.dumps(record),
        )
        # inserted is None on conflict — the record was already committed by a
        # prior delivery, so the terminal state is ACCEPTED either way.
        state = TerminalState.ACCEPTED
        # Append-only audit row: who/when/what for every transition.
        await conn.execute(
            "INSERT INTO permit_audit (idempotency_key, state, worker, ts) "
            "VALUES ($1, $2, $3, now())",
            record["idempotency_key"], state, WORKER_ID,
        )
        return state

Every state change — received, chunked, enriched, validated, accepted, rejected, quarantined — is written to the append-only permit_audit table with a timestamp, worker identifier, and idempotency key. That table is the artifact compliance officers query during public-records requests and state audits; it guarantees no submission ever silently disappears.

Configuration Reference

Parameter Type Default Municipal-context notes
MAX_RECORDS int 500 Records per batch. Lower for heavy enrichment; align with reconciliation reporting units.
MAX_BYTES int 25 MB Payload ceiling per batch. Keep under the object-store multipart threshold to avoid split uploads.
MAX_CONCURRENCY int 32 In-flight records per worker. Never exceed the database pool size minus headroom for interactive queries.
prefetch_count (QoS) int 4 Batches the broker hands one worker at a time. Caps memory during surges; second throttle behind the semaphore.
pool_max_size int 20 asyncpg connections per worker. Shared by all coroutines; size against total worker count, not per-coroutine.
enrichment_timeout float 5.0 Per-lookup ceiling (seconds) on registry calls. Trips a transient retry, not a permanent rejection.
idempotency_ttl int 86400 Redis TTL (seconds) for dedup keys; match the source system’s reconciliation window.
priority str standard Routing key tier — critical (zoning/environmental), standard, or maintenance (data hygiene).

Error Handling and Edge Cases

Municipal data is messy, and async pipelines surface failure modes that synchronous ones hide. Handle these explicitly:

  • Transient vs. permanent faults. Wrap broker and database I/O in a retry policy with exponential backoff and jitter so a brief network partition recovers automatically; route schema-validation failures straight to the dead-letter queue with no retry. Jitter prevents synchronized retry storms when several workers restart together after an outage.
from tenacity import retry, stop_after_attempt, wait_exponential_jitter


@retry(stop=stop_after_attempt(4), wait=wait_exponential_jitter(initial=0.5, max=30))
async def fetch_zoning(client, parcel_id: str) -> str:
    # Retries only on transient transport errors; a 4xx schema error should
    # raise a non-retryable exception so the record is quarantined, not looped.
    resp = await client.get(f"/zoning/{parcel_id}", timeout=5.0)
    resp.raise_for_status()
    return resp.json()["zoning_code"]
  • Poison messages. A batch that fails deserialization every time will block the queue head if it is endlessly requeued. Cap redeliveries with x-death header inspection and divert to the dead-letter queue after the threshold so one malformed file never stalls overnight processing.
  • Partial-batch crashes. Because each record commits in its own transaction and is keyed by idempotency hash, a worker that dies mid-batch simply lets the broker redeliver the envelope; already-committed records hit the ON CONFLICT no-op and only the unfinished ones are reprocessed.
  • Encoding and delimiter drift in source files. Legacy SFTP exports arrive in mixed encodings and inconsistent delimiters; normalize and validate file integrity on receipt, quarantining malformed payloads. The relational-mapping side of that work lives in Syncing Legacy CSV Exports to Modern Databases.
  • Connection-pool exhaustion. If MAX_CONCURRENCY exceeds pool_max_size, coroutines block waiting for a connection and latency balloons silently. Keep the semaphore bound at or below the pool size and alert on pool-wait time.

Testing and Verification

Confidence in an async pipeline comes from simulating the failures that production will throw at it. Use pytest-asyncio with an in-memory broker stub and a transactional test database.

import pytest


@pytest.mark.asyncio
async def test_redelivered_batch_is_idempotent(pool) -> None:
    record = {"idempotency_key": "abc123", "submission_id": "P-1001", "x": 1}
    # First delivery commits; second (redelivery) must be a no-op.
    async with pool.acquire() as conn:
        first = await commit(record, conn)
        second = await commit(record, conn)
    assert first is TerminalState.ACCEPTED and second is TerminalState.ACCEPTED
    async with pool.acquire() as conn:
        count = await conn.fetchval(
            "SELECT count(*) FROM permits WHERE idempotency_key = $1", "abc123"
        )
    assert count == 1  # exactly-once despite two deliveries


@pytest.mark.asyncio
async def test_invalid_submission_is_quarantined() -> None:
    with pytest.raises(ValidationError):
        PermitSubmission.model_validate({"submission_id": "P-2002"})  # missing fields

Beyond unit tests, assert operational invariants after a load run: every batch’s record count reconciles against committed-plus-quarantined totals, no row sits in a non-terminal state, and the audit table holds one transition chain per idempotency key. A chaos test that kills a worker mid-batch and confirms zero duplicate commits is the single most valuable check you can keep in CI.

Integration Notes

This pipeline is the throughput layer the rest of the permitting stack feeds into and out of.

For event-loop internals and worker lifecycle specifics, the official Python asyncio documentation and Celery distributed task guide remain the authoritative references.

Frequently Asked Questions

When should I choose asyncio over Celery for permit batch jobs?

Reach for plain asyncio when the work is I/O-bound — registry lookups, broker and database calls — and you want one process to drive thousands of concurrent coroutines cheaply. Choose Celery when you need durable scheduled dispatch, multi-host worker fan-out, and built-in retry/visibility tooling for overnight windows. Many municipal stacks run both: Celery owns scheduling and worker lifecycle, while each task uses asyncio internally for its concurrent lookups.

How do I guarantee a permit is processed exactly once?

Derive a deterministic idempotency key from the payload digest, enforce it with a unique database constraint, and commit each record with ON CONFLICT DO NOTHING. A redelivered message then hits the constraint and becomes a no-op, so no duplicate record or fee is created even when the broker delivers the same batch twice.

What batch size and concurrency should I start with?

Begin at 500 records per batch and 32 in-flight records per worker, then tune against your database connection pool — concurrency must stay at or below the pool size minus headroom for interactive clerk queries. Scale total throughput by adding worker processes, not by pushing per-process concurrency past what the database can absorb.

What happens to submissions that fail validation?

They are routed to a dead-letter (quarantine) queue with their error signature and never retried, because a missing statutory field or failed zoning cross-reference is a permanent fault. Compliance officers review the quarantine queue; the audit table records the quarantined transition so nothing is silently dropped.

Explore deeper

Sub-topic Choosing Celery vs asyncio for Permit Batch Jobs Sub-topic Managing Celery Task Queues for Overnight Batch Imports