Async Batch Processing for Municipal GIS Ingestion

Municipal zoning amendments and parcel boundary updates rarely arrive as single, atomic transactions. PropTech platforms and planning departments ingest tens of thousands of spatial records per day from fragmented county portals, planning-commission minutes, and aging GIS servers — and the moment a single synchronous loop tries to download, reproject, validate, and write every feature in sequence, throughput collapses and county servers start returning 429s. Async batch processing exists to solve exactly that failure mode: it keeps the event loop saturated with concurrent network requests while heavy geometry work runs off-loop, so a county-wide refresh that took six hours of blocking I/O finishes in minutes without tripping rate limits. This page is part of the Automated Feed Ingestion & GIS Data Parsing framework, and it sits downstream of acquisition and upstream of export — the engine that turns a backlog of raw feeds into a steady, deterministic stream of validated parcels.

Prerequisites and operational context jump to heading

Async batch processing is not the first thing you build — it is the thing you build once the surrounding pieces exist. Several conditions must already be in place before the patterns below pay off:

• A staging layer. Raw payloads should land in versioned object storage (S3/GCS with object versioning, or a raw_features table) before any transformation. The batch processor reads from staging, never directly from the live municipal endpoint, so that a reprocessing run never re-hammers a county server.
• A known target projection. Every feature is reprojected to a single working CRS during the batch. That depends on the CRS alignment strategies you have standardised on — typically EPSG:4326 for storage and EPSG:3857 for tiling — so that geometry validation and spatial joins operate on consistent coordinates.
• A rate-limit budget. You must know each source’s throttling behaviour before you set concurrency. Concurrency limits are derived from municipal API rate limit management policy, not guessed.
• A broker or work queue. For anything beyond a single machine, partitions are enqueued in Redis Streams, RabbitMQ, or SQS so that workers can be added horizontally and failed partitions can be replayed.
• A normalization contract. The attribute keys the processor writes (parcel_id, zoning_code, jurisdiction) must already conform to your attribute normalization rules, otherwise the batch silently writes inconsistent zoning codes across jurisdictions.

If any of these are missing, fix them first. Async concurrency amplifies whatever you feed it — including malformed input and unbounded request rates.

Architecture: partitioning and the producer-consumer split jump to heading

Effective batching begins with intelligent partitioning. Municipal datasets often span entire counties or metropolitan statistical areas, making monolithic downloads impractical and prone to timeout. Partition feeds along a stable, deterministic key so that the same chunk always maps to the same work unit:

• By jurisdiction — one partition per municipality, which aligns naturally with how rate limits and schemas vary.
• By zoning district or land-use code — useful when a single large city publishes overlays separately.
• By spatial index tile — H3 hexagons or a quadtree grid, which keeps each chunk’s geometry count bounded regardless of how dense the parcels are.

Each partition becomes an independent async task with its own retry state, so one county’s outage never blocks another’s progress. The system has two clearly separated halves. The producer reads from upstream queues — for jurisdictions without an OGC API, that upstream is the output of the PDF & HTML scraping pipelines — assigns chunk IDs, and pushes work onto the broker. The consumer pulls chunks, fetches features over HTTP, reprojects and validates geometry, and writes idempotent upserts.

The single most important design decision is the concurrency model. Network fetches are I/O-bound and belong on the event loop; geometry validation is CPU-bound and must not run there. Python’s Global Interpreter Lock means a make_valid or intersection call on a complex multipolygon will freeze the entire loop if you await it inline. The pattern that scales:

• Bound HTTP concurrency with an asyncio.Semaphore (typically 15–30 in flight) so you never exceed the county’s tolerated request rate.
• Off-load every GEOS-backed operation to a ThreadPoolExecutor (Shapely 2.x releases the GIL during GEOS calls) or, for very heavy validation, a ProcessPoolExecutor.
• Keep JSON parsing, connection pooling, and database cursors non-blocking so the loop stays responsive while geometry work happens elsewhere.

This separation — bounded async I/O in front, a worker pool behind — is what lets a single process sustain high throughput without either starving the loop or melting a municipal server.

Production implementation jump to heading

The following processor is tailored for municipal zoning feeds. It applies bounded concurrency, retry with jittered backoff, thread-pool geometry validation, and transactional upserts. The code is complete and runnable against an aiohttp session and an async upsert callable.

import asyncio
import logging
import random
from concurrent.futures import ThreadPoolExecutor
from typing import Callable, Dict, List, Optional, Tuple

import aiohttp
from shapely.geometry import shape
from shapely.validation import make_valid

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
)
logger = logging.getLogger(__name__)

# --- Concurrency and resource boundaries -------------------------------------
# Cap concurrent HTTP requests so we never exceed the county's tolerated rate.
API_SEMAPHORE = asyncio.Semaphore(20)
MAX_RETRIES = 3
BASE_DELAY = 1.0
# Shapely 2.x releases the GIL inside GEOS, so a thread pool gives real
# parallelism for geometry validation without the IPC cost of processes.
GEOS_POOL = ThreadPoolExecutor(max_workers=8)
RETRYABLE_STATUS = {429, 502, 503, 504}


async def backoff(attempt: int, retry_after: Optional[float] = None) -> None:
    """Exponential backoff with full jitter; honour Retry-After when present."""
    if retry_after is not None:
        await asyncio.sleep(retry_after)
        return
    delay = BASE_DELAY * (2 ** attempt) + random.uniform(0.0, 0.5)
    await asyncio.sleep(delay)


def validate_geometry(feature: Dict) -> Optional[str]:
    """CPU-bound geometry validation, run inside the thread pool.

    Returns repaired WKT, or None for empty/invalid geometry so the caller
    can route the record to quarantine instead of writing a bad polygon.
    """
    try:
        geom = shape(feature.get("geometry", {}))
        if geom.is_empty:
            return None
        repaired = make_valid(geom)
        return repaired.wkt if not repaired.is_empty else None
    except Exception as exc:  # malformed coordinates, unsupported type, etc.
        logger.warning("Geometry validation failed: %s", exc)
        return None


async def fetch_chunk(
    session: aiohttp.ClientSession,
    chunk_id: str,
    api_endpoint: str,
) -> Optional[Dict]:
    """Fetch one spatial chunk with bounded concurrency and retry."""
    for attempt in range(MAX_RETRIES):
        try:
            async with API_SEMAPHORE:
                async with session.get(api_endpoint, params={"chunk": chunk_id}) as resp:
                    if resp.status in RETRYABLE_STATUS and attempt < MAX_RETRIES - 1:
                        retry_after = resp.headers.get("Retry-After")
                        await backoff(attempt, float(retry_after) if retry_after else None)
                        continue
                    resp.raise_for_status()
                    return await resp.json()
        except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
            if attempt < MAX_RETRIES - 1:
                await backoff(attempt)
                continue
            logger.error("Chunk %s failed after %d attempts: %s", chunk_id, MAX_RETRIES, exc)
    return None


async def process_chunk(
    session: aiohttp.ClientSession,
    chunk_id: str,
    api_endpoint: str,
    upsert: Callable[[List[Dict]], "asyncio.Future"],
) -> Dict[str, int]:
    """Fetch -> validate (off-loop) -> idempotent upsert for one chunk."""
    payload = await fetch_chunk(session, chunk_id, api_endpoint)
    if payload is None:
        return {"chunk_id": chunk_id, "success": 0, "errors": 1, "quarantined": 0}

    features = payload.get("features", [])
    if not features:
        return {"chunk_id": chunk_id, "success": 0, "errors": 0, "quarantined": 0}

    # Off-load every GEOS call to the thread pool; the event loop stays free.
    loop = asyncio.get_running_loop()
    validated = await asyncio.gather(
        *(loop.run_in_executor(GEOS_POOL, validate_geometry, feat) for feat in features)
    )

    records, quarantined = [], 0
    for feat, wkt in zip(features, validated):
        if wkt is None:
            quarantined += 1
            continue
        props = feat.get("properties", {})
        records.append({
            "parcel_id": props.get("parcel_id"),
            "zoning_code": props.get("zoning"),
            "geometry_wkt": wkt,
            "source_chunk": chunk_id,
            "last_updated": payload.get("timestamp"),
        })

    if not records:
        return {"chunk_id": chunk_id, "success": 0, "errors": 0, "quarantined": quarantined}

    try:
        await upsert(records)  # transactional INSERT ... ON CONFLICT DO UPDATE
        return {"chunk_id": chunk_id, "success": len(records), "errors": 0,
                "quarantined": quarantined}
    except Exception as exc:
        logger.error("Upsert failed for chunk %s: %s", chunk_id, exc)
        return {"chunk_id": chunk_id, "success": 0, "errors": len(records),
                "quarantined": quarantined}


async def run_pipeline(
    chunks: List[Tuple[str, str]],
    upsert: Callable[[List[Dict]], "asyncio.Future"],
) -> Dict[str, int]:
    """Orchestrate the batch; one task per partition, results aggregated."""
    timeout = aiohttp.ClientTimeout(total=120, connect=10)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        results = await asyncio.gather(
            *(process_chunk(session, cid, url, upsert) for cid, url in chunks),
            return_exceptions=True,
        )

    totals = {"success": 0, "errors": 0, "quarantined": 0}
    for r in results:
        if isinstance(r, dict):
            for k in totals:
                totals[k] += r.get(k, 0)
        else:  # an exception escaped a task
            totals["errors"] += 1
            logger.error("Unhandled task error: %s", r)

    logger.info("Pipeline complete: %s", totals)
    return totals

A few choices in this implementation are deliberate. fetch_chunk honours the Retry-After header before falling back to computed backoff, which matters because many municipal gateways return an explicit cooldown. Validation failures return None rather than raising, so a single bad polygon quarantines one record instead of failing an entire chunk. And run_pipeline uses return_exceptions=True so that one task crashing never cancels the siblings already in flight.

Edge cases and gotchas jump to heading

Municipal feeds break in specific, recurring ways. Plan for these before they reach production:

• 429 storms across partitions. A global Semaphore of 20 still means 20 simultaneous requests to the same county if all your partitions belong to one jurisdiction. Use per-host semaphores or a per-jurisdiction token bucket so one slow county cannot consume the whole budget.
• Topology exceptions on repair. make_valid can return a GeometryCollection (e.g. a polygon plus a stray line) when it fixes a bowtie. Writing that to a Polygon/MultiPolygon column raises a type error downstream. Filter the collection to its polygonal parts, or quarantine it.
• Datum drift hidden in metadata. A feed may declare EPSG:4326 in its envelope but ship coordinates in a state plane system. Validation passes, but the parcel lands in the ocean. Sanity-check that transformed bounds fall inside the jurisdiction’s expected bounding box.
• Schema version mismatches. When a county renames zoning to zone_class between releases, props.get("zoning") silently returns None. Assert on required keys and route misses to quarantine rather than writing null zoning codes.
• Unbounded thread-pool growth. Submitting tens of thousands of run_in_executor calls at once queues them faster than eight workers can drain. Chunk the gather, or gate executor submissions, so memory stays flat.
• Event-loop starvation from accidental sync calls. A single blocking requests.get or a synchronous DB driver inside an async task freezes everything. Keep the hot path strictly async; this is where error handling & retry logic belongs — scoped per stage, not bolted on globally.

Retry strategy reference jump to heading

Different failure classes deserve different retry behaviour. Treating them uniformly either wastes compute or masks real data problems.

Integration points jump to heading

Async batch processing is a midstream stage, defined by what feeds it and what consumes it. Its inputs are the staged payloads produced by the scraping and acquisition layers; its outputs flow forward to publication. Once geometries are validated and attributes normalized, records become the input to GIS export sync workflows, which serialise them back out as shapefiles, GeoJSON, or vector tiles for dashboards, compliance reporting, and public parcel viewers. When a partition cannot be processed after exhausting retries, it should hand off to fallback routing logic so that a missing feed is filled from a secondary source or last-known-good snapshot rather than leaving a gap in the map. The contract in both directions is the normalized record shape — same keys, same CRS, same quarantine semantics — so that batching can be swapped, scaled, or replayed without downstream changes.

Compliance and audit artifacts jump to heading

For PropTech underwriting and regulatory review, a batch run is not done when the data lands — it is done when the run is provable. Every batch should emit:

• A run manifest: start/end timestamps, source endpoints, the working CRS, library versions (Shapely/GEOS, pyproj), and the partition list. This is the data-lineage record that links each parcel back to the exact feed and code that produced it.
• Per-chunk counts: fetched, validated, quarantined, and upserted, so a reviewer can reconcile input volume against what reached the database.
• Deterministic content hashes: hash each chunk’s payload so re-runs are idempotent and any silent change in upstream data is detectable.
• A quarantine ledger: every rejected record with its reason (invalid geometry, schema miss, datum check failure), retained immutably for audit.

These artifacts pair with downstream schema validation & data quality checks to form a continuous, defensible chain from raw municipal publication to underwriting-grade dataset.

FAQ jump to heading

Related jump to heading

• Implementing async batch processing for large GIS datasets — memory profiling, chunk sizing, and backpressure in depth
• Municipal API rate limit management — sets the concurrency budget this engine consumes
• Error handling & retry logic — per-stage failure boundaries and recovery
• GIS export sync workflows — the downstream consumer of validated records
• CRS alignment strategies — the projection contract every batch enforces

Up: Automated Feed Ingestion & GIS Data Parsing