Async Bulk Geospatial Uploads with Celery

← Back to Advanced Spatial Endpoints & Data Contracts

Processing large spatial datasets synchronously blocks your request lifecycle and guarantees timeouts above a few hundred features. When your API needs to ingest shapefiles, GeoJSON archives, or CSV coordinate dumps at scale, you need a queued pipeline that decouples file acceptance from geometry parsing, CRS transformation, and PostGIS writes. This page walks through production-ready patterns for building that pipeline with Celery and FastAPI — covering broker configuration, idempotent batch insertion, worker reliability settings, and client-facing job tracking.

# Prerequisites & Environment

Broker choice: Redis is simpler to operate and sufficient for most spatial ingestion workloads. RabbitMQ is preferable when you need strong per-message durability guarantees, complex routing topologies, or reliable dead-letter queues for failed shapefile jobs. For result_expires, set 86400 (24 hours) on both backends to prevent job-metadata bloat.

# Pipeline Architecture

The diagram below shows how a raw client upload flows through FastAPI into Celery workers and finally into PostGIS.

Each stage is intentionally isolated: the FastAPI endpoint never opens a database connection during upload, and Celery workers use a separate synchronous connection pool (psycopg2) independent of the async pool (asyncpg) that serves read queries.

# Decision Matrix: Broker & Storage Trade-offs

For multi-node worker pools, object storage staging is the correct default. Workers download the file at the start of the task and delete it on completion.

# Step-by-Step Implementation

# Step 1 — FastAPI ingestion endpoint

The endpoint’s only job is to accept the file, write it to staging, and enqueue a task. It must return before any geometry parsing occurs. The handling async file uploads for shapefile processing page covers companion-file validation (.prj, .dbf, .cpg) and multipart ZIP extraction in detail.

from fastapi import APIRouter, UploadFile, HTTPException
from uuid import uuid4
import os

router = APIRouter()
UPLOAD_DIR = "/var/geospatial/staging"
os.makedirs(UPLOAD_DIR, exist_ok=True)

ALLOWED_EXTENSIONS = (".zip", ".geojson", ".gpkg", ".csv")

@router.post("/v1/uploads/geospatial", status_code=202)
async def upload_geospatial(file: UploadFile):
    if not file.filename:
        raise HTTPException(400, "No filename provided")
    ext = os.path.splitext(file.filename)[1].lower()
    if ext not in ALLOWED_EXTENSIONS:
        raise HTTPException(400, f"Unsupported format: {ext!r}. Expected one of {ALLOWED_EXTENSIONS}")

    job_id = str(uuid4())
    staging_path = os.path.join(UPLOAD_DIR, f"{job_id}{ext}")

    # Stream in 1 MB chunks — never load the full payload into RAM
    with open(staging_path, "wb") as f:
        while chunk := await file.read(1_048_576):
            f.write(chunk)

    # Dispatch immediately; do not await the result
    from celery_app import process_geospatial_upload
    process_geospatial_upload.delay(job_id=job_id, file_path=staging_path)

    return {
        "job_id": job_id,
        "status": "queued",
        "poll_url": f"/v1/jobs/{job_id}",
    }

Returning 202 Accepted — not 200 OK — is the correct HTTP semantic for queued work. Clients must not assume the features are immediately queryable after receiving this response.

# Step 2 — Celery application configuration

Worker reliability for geospatial workloads depends on three settings: task_acks_late, worker_prefetch_multiplier, and worker_max_tasks_per_child. Missing any one of them causes silent job loss or memory exhaustion over time.

from celery import Celery

app = Celery(
    "geospatial_worker",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)

app.conf.update(
    # Acknowledge only after the task function returns — crash-safe
    task_acks_late=True,
    # Process one task at a time; prevents memory spikes from concurrent GDAL ops
    worker_prefetch_multiplier=1,
    # Recycle the worker process every 50 tasks — counters GDAL/shapely memory fragmentation
    worker_max_tasks_per_child=50,
    # Keep serialization deterministic for spatial payloads
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    # Expire result metadata after 24 hours to prevent Redis bloat
    result_expires=86_400,
)

task_acks_late=True is the most critical setting for ingestion pipelines: without it, the broker marks a task as acknowledged the moment a worker picks it up. If the worker process is killed mid-parse, the geometry records are silently lost with no retry.

# Step 3 — Worker task: extract, validate, and transform

Geometry extraction, validation, and CRS transformation are CPU-bound and may allocate several hundred MB of C-level memory for large shapefiles. Isolating this inside a Celery task prevents the FastAPI event loop from blocking and allows the OS to reclaim memory when the worker process recycles.

import logging
import pyogrio
import shapely
from shapely.validation import make_valid
from pyproj import Transformer

logger = logging.getLogger(__name__)

@app.task(bind=True, max_retries=3, default_retry_delay=60)
def process_geospatial_upload(self, job_id: str, file_path: str):
    try:
        _update_job_status(job_id, "processing", progress=0)

        # --- 1. Read metadata without loading all features ---
        meta = pyogrio.read_info(file_path)
        src_crs = meta.get("crs")
        if src_crs is None:
            raise ValueError("Source file has no CRS; cannot transform to EPSG:4326")

        # --- 2. Stream features and transform coordinates ---
        transformer = Transformer.from_crs(src_crs, "EPSG:4326", always_xy=True)
        gdf = pyogrio.read_dataframe(file_path)

        geometries = []
        for i, geom in enumerate(gdf.geometry):
            if geom is None:
                continue
            reprojected = shapely.transform(geom, transformer.transform)
            if not reprojected.is_valid:
                reprojected = make_valid(reprojected)
            geometries.append((job_id, i, reprojected))

        _update_job_status(job_id, "processing", progress=60)

        # --- 3. Batch insert ---
        batch_insert_geometries(geometries, table_name="spatial_features", db_url=DB_URL)

        _update_job_status(job_id, "completed", progress=100, record_count=len(geometries))
        logger.info("Job %s completed: %d features inserted", job_id, len(geometries))

    except Exception as exc:
        logger.exception("Job %s failed on attempt %d", job_id, self.request.retries + 1)
        _update_job_status(job_id, "failed", error=str(exc))
        raise self.retry(exc=exc)

always_xy=True on the Transformer is not optional — omitting it causes latitude and longitude to swap for any CRS that uses the geographic (lat/lon) axis order, silently producing mirror-image geometries.

# Step 4 — Idempotent batch insertion with ON CONFLICT

Using psycopg2.extras.execute_values with page_size=1000 and an ON CONFLICT DO NOTHING clause makes each task retry safe. If the worker crashes after 700 rows and retries from the top, the 700 already-committed rows are skipped rather than duplicated. This is the core of the strict data contract that governs how you enforce idempotency at the database layer.

import psycopg2
from psycopg2.extras import execute_values

def batch_insert_geometries(geometries: list[tuple], table_name: str, db_url: str):
    """
    geometries: list of (job_id, row_index, shapely_geom) tuples
    Uses (job_id, row_index) as a composite idempotency key.
    """
    conn = psycopg2.connect(db_url)
    try:
        with conn.cursor() as cur:
            execute_values(
                cur,
                f"""
                INSERT INTO {table_name} (job_id, row_index, geometry, created_at)
                VALUES %s
                ON CONFLICT (job_id, row_index) DO NOTHING
                """,
                [
                    (job_id, row_idx, geom.wkt)
                    for job_id, row_idx, geom in geometries
                ],
                template="(%s, %s, ST_GeomFromText(%s, 4326), NOW())",
                page_size=1000,
            )
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

The page_size=1000 parameter controls how many rows are bundled into a single VALUES clause. Values above ~2000 rows per batch begin to stress the PostgreSQL query planner without meaningful throughput gains.

# Production Code Example

The following is a complete, copy-runnable integration showing the FastAPI status endpoint, the Celery task wired to the batch insert, and the helper that persists job state to Redis. Drop these three modules into your project alongside the snippets above.

# status_store.py  — Redis-backed job state
import json
import redis

r = redis.Redis(host="localhost", port=6379, db=2, decode_responses=True)

def _update_job_status(
    job_id: str,
    status: str,
    progress: int = 0,
    record_count: int | None = None,
    error: str | None = None,
):
    payload = {"job_id": job_id, "status": status, "progress": progress}
    if record_count is not None:
        payload["record_count"] = record_count
    if error is not None:
        payload["error"] = error
    r.setex(f"job:{job_id}", 86_400, json.dumps(payload))

def get_job_status(job_id: str) -> dict | None:
    raw = r.get(f"job:{job_id}")
    return json.loads(raw) if raw else None

# routes/jobs.py  — polling endpoint
from fastapi import APIRouter, HTTPException
from status_store import get_job_status

router = APIRouter()

@router.get("/v1/jobs/{job_id}")
async def job_status(job_id: str):
    state = get_job_status(job_id)
    if state is None:
        raise HTTPException(404, f"Job {job_id!r} not found or expired")
    return state

A client polling every 2 seconds with exponential backoff can track progress without long-polling infrastructure. When status == "completed", it can immediately run bounding-box spatial index queries against the newly ingested features, or issue K-nearest-neighbor routing queries against the populated geometry table.

# Verification & Testing

1. Smoke test with curl:

# Upload a sample GeoJSON
curl -X POST http://localhost:8000/v1/uploads/geospatial \
  -F "file=@sample_features.geojson" \
  -H "Accept: application/json"
# → {"job_id":"a1b2c3...","status":"queued","poll_url":"/v1/jobs/a1b2c3..."}

# Poll until completed
curl http://localhost:8000/v1/jobs/a1b2c3...
# → {"job_id":"a1b2c3...","status":"completed","progress":100,"record_count":1458}

2. Verify geometry quality in PostGIS:

-- Confirm all inserted geometries are valid and in EPSG:4326
SELECT
    COUNT(*) AS total,
    COUNT(*) FILTER (WHERE ST_IsValid(geometry)) AS valid_count,
    COUNT(*) FILTER (WHERE ST_SRID(geometry) = 4326) AS correct_srid,
    MIN(ST_NPoints(geometry)) AS min_vertices,
    MAX(ST_NPoints(geometry)) AS max_vertices
FROM spatial_features
WHERE job_id = 'a1b2c3...';

Expected output: total == valid_count == correct_srid.

3. Confirm idempotency (re-run same task, expect same count):

# tests/test_idempotency.py
import pytest
from celery_app import process_geospatial_upload

def test_duplicate_task_is_idempotent(db_conn, sample_shapefile):
    job_id = "test-idem-001"
    process_geospatial_upload(job_id=job_id, file_path=sample_shapefile)
    count_after_first = db_conn.execute(
        "SELECT COUNT(*) FROM spatial_features WHERE job_id = %s", (job_id,)
    ).fetchone()[0]

    # Run again — must not double-insert
    process_geospatial_upload(job_id=job_id, file_path=sample_shapefile)
    count_after_second = db_conn.execute(
        "SELECT COUNT(*) FROM spatial_features WHERE job_id = %s", (job_id,)
    ).fetchone()[0]

    assert count_after_first == count_after_second

# Failure Modes & Edge Cases

1. Missing CRS in shapefile — pyogrio.read_info() returns crs: None when a .prj companion file is absent. The task raises ValueError and retries. After max_retries, the job is marked failed. Fix: reject uploads lacking .prj at the FastAPI layer before staging; see handling async file uploads for shapefile processing.

2. Lat/lon axis swap producing ocean-based geometries — Omitting always_xy=True from Transformer.from_crs() swaps axes for CRS definitions that declare (latitude, longitude) order (e.g. EPSG:4269, EPSG:4326 in strict mode). Symptoms: all geometries land in the ocean or Antarctica. Always pass always_xy=True.

3. Worker OOM kill during large archive processing — A 500 MB shapefile with complex polygons can exhaust 4 GB of worker RAM when loaded fully into a GeoDataFrame. Mitigate with pyogrio.read_dataframe() chunked reads (use the skip_features/max_features parameters in a loop) or stream via Fiona’s iterative layer reading.

4. Transaction rollback leaving partial job state — If the database transaction rolls back after 3,000 of 10,000 rows, the job status in Redis may already read processing. On retry, the ON CONFLICT DO NOTHING clause skips the 3,000 already-committed rows (if you committed partials). Use a single transaction per batch, not per page: commit only after all execute_values pages succeed.

5. Redis TTL expiry before client polls — result_expires=86_400 means job metadata disappears after 24 hours. If a client polls after that window, it receives a 404. Persist completed job summaries to a job_log Postgres table before they expire in Redis if you need longer audit trails.

6. GiST index bloat after bulk insert — Inserting millions of geometries in rapid succession fragments the GiST index. Query plans for bounding-box queries using ST_Within and ST_Intersects may degrade significantly. Schedule VACUUM ANALYZE spatial_features and REINDEX INDEX spatial_features_geometry_idx in a maintenance window after each large bulk load.

7. Celery task swallowing exceptions silently — If self.retry(exc=exc) is called inside a bare except Exception block without re-raising, Celery may mark the task SUCCESS after the retry limit is reached. Always use raise self.retry(exc=exc) to propagate correctly.

# Performance Notes

For very large loads (>500,000 features), consider dropping and recreating the GiST index around the bulk operation rather than inserting into a live index:

-- Before bulk load
DROP INDEX IF EXISTS spatial_features_geometry_idx;

-- ... run Celery tasks ...

-- After all tasks complete
CREATE INDEX CONCURRENTLY spatial_features_geometry_idx
    ON spatial_features USING GIST (geometry);
VACUUM ANALYZE spatial_features;

CREATE INDEX CONCURRENTLY avoids a full table lock but takes longer. Use it in production where you cannot afford downtime on spatial queries. This intersects with the connection pool tuning discussed in Connection Pooling & pgBouncer Setup — ensure your pgBouncer pool size accommodates the CONCURRENTLY build’s additional session.

# Related

• Handling Async File Uploads for Shapefile Processing — companion-file validation, ZIP extraction, and chunked multipart parsing
• Strict Pydantic Validation for Geometry — enforce geometry types and CRS contracts at the request boundary before queueing
• Bounding Box & Spatial Index Queries — query the PostGIS tables populated by this pipeline
• K-Nearest Neighbor Routing Algorithms — KNN queries over bulk-inserted geometry datasets
• Connection Pooling & pgBouncer Setup — tune database connection pools for high-concurrency bulk ingest

← Back to Advanced Spatial Endpoints & Data Contracts