Design a Geospatial Webhook Architecture in Python

To design a geospatial webhook architecture in Python, decouple spatial event ingestion from HTTP delivery using an async message broker, validate geometry with Shapely before dispatch, sign every payload with HMAC-SHA256, and apply exponential backoff with dead-letter routing for undeliverable events.

This page is part of the Feature Change Triggers topic under Core Event Fundamentals & Architecture. It focuses on the single concrete question of how to wire these four concerns together into a runnable Python service.

When to use this pattern

Use the four-layer async architecture described here when:

  • Your spatial data source emits high-frequency feature mutations (PostGIS triggers, GDAL pipelines, IoT sensor feeds) and a synchronous HTTP call on every write would introduce unacceptable latency or backpressure.
  • Consumers need guaranteed at-least-once delivery — a fire-and-forget HTTP call on a database trigger is not sufficient, because you lose the event if the consumer is temporarily unavailable.
  • Payload validation is non-trivial: you need to check geometry topology, enforce coordinate ring orientation, or reject coordinates outside WGS 84 (EPSG:4326) bounds before events leave your system.

If you only need to push a single event type to a single internal service with no retry requirements, a simpler synchronous requests.post is fine. This architecture pays off when you operate across multiple tenants, multiple endpoint registrations, or geographies with variable network reliability.

Architecture overview

The pipeline operates across four logical layers. Each isolates a single concern so that failures do not cascade and each layer can scale independently.

Four-layer geospatial webhook pipeline Data flows left-to-right from the Ingestion layer through Spatial Validation, then a Message Broker, to the Async Dispatcher which delivers signed payloads to consumer endpoints. 1. Ingestion 2. Validation 3. Broker 4. Dispatcher PostGIS trigger GDAL pipeline IoT sensor feed Client SDK event Shapely topology CRS → EPSG:4326 Pydantic schema Delta threshold Redis Streams Topic per feature Subscription index Ordered by feat. ID HMAC-SHA256 sign aiohttp async POST Exponential backoff Dead-letter queue DLQ replay path

Layer 1 — Ingestion: Receives raw feature mutations and normalizes them into a canonical schema: event_id (UUID), feature_id, geometry (GeoJSON dict), change_type (create | update | delete), and properties. This layer is write-optimized and never blocks downstream workers.

Layer 2 — Spatial Validation: Parses geometry through Shapely, checks topology validity, and evaluates spatial delta thresholds — the same logic that drives Feature Change Triggers. Events that do not cross a meaningful delta (micro-edits, null-geometry noise) are dropped here, not forwarded.

Layer 3 — Message Broker: Publishes validated events to Redis Streams, RabbitMQ, or Kafka. Subscriber matching happens via spatial bounding-box indexes or attribute tags. The broker guarantees ordering per feature_id and persists events until acknowledged. Storing idempotency keys in the broker maps directly to the Event Key Generation for Spatial Data strategy.

Layer 4 — Async Dispatcher: Pulls events, signs payloads with HMAC-SHA256 (as detailed in Securing Webhook Endpoints with Spatial Token Validation), and delivers them over non-blocking HTTP. Delivery state, retry counts, and consumer health are tracked in a sidecar store.

Complete runnable implementation

The module below is self-contained and production-aligned. It wires all four layers: Pydantic v2 schema with Shapely validation, HMAC signing, async delivery with jitter, and dead-letter routing. No placeholder TODOs — every part runs as written.

python 
# geospatial_webhook_dispatcher.py
#
# Dependencies:
#   pip install fastapi aiohttp pydantic shapely uvicorn
#
# Run with:
#   uvicorn geospatial_webhook_dispatcher:app --host 0.0.0.0 --port 8000

import asyncio
import hashlib
import hmac
import json
import logging
import random
import uuid
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from typing import Literal

import aiohttp
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel, Field, field_validator
from shapely.geometry import shape
from shapely.validation import explain_validity

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

# One shared aiohttp.ClientSession per worker process — created on startup,
# closed on shutdown via the lifespan handler defined below.
_session: aiohttp.ClientSession | None = None


@asynccontextmanager
async def lifespan(_: FastAPI):
    # Reuse TCP connections + DNS cache across every POST so retries do not
    # repeat the TLS handshake. on_event("startup"/"shutdown") is deprecated
    # in modern FastAPI; the lifespan context manager is the supported path.
    global _session
    connector = aiohttp.TCPConnector(limit=50, ttl_dns_cache=300)
    _session = aiohttp.ClientSession(connector=connector)
    try:
        yield
    finally:
        await _session.close()


app = FastAPI(title="Geospatial Webhook Dispatcher", lifespan=lifespan)

# ---------------------------------------------------------------------------
# 1. Canonical GeoEvent schema
#    All fields follow RFC 7946 conventions; geometry must be WGS 84 (EPSG:4326).
# ---------------------------------------------------------------------------

class GeoEvent(BaseModel):
    event_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    feature_id: str
    geometry: dict                              # RFC 7946 GeoJSON geometry object
    change_type: Literal["create", "update", "delete"]
    properties: dict = Field(default_factory=dict)
    timestamp: datetime = Field(
        default_factory=lambda: datetime.now(timezone.utc)
    )

    @field_validator("geometry")
    @classmethod
    def validate_geojson_geometry(cls, v: dict) -> dict:
        """
        Parse through Shapely to catch topology errors before the event
        reaches the broker.  Self-intersecting polygons, unclosed rings, and
        empty geometries all raise ValueError here, not at delivery time.
        """
        if not v or "type" not in v or "coordinates" not in v:
            raise ValueError("geometry must be a valid GeoJSON geometry object")
        try:
            geom = shape(v)
        except Exception as exc:
            raise ValueError(f"Could not parse geometry: {exc}") from exc

        if geom.is_empty:
            raise ValueError("geometry must not be empty")

        if not geom.is_valid:
            detail = explain_validity(geom)
            raise ValueError(f"Invalid topology: {detail}")

        return v


# ---------------------------------------------------------------------------
# 2. HMAC-SHA256 payload signing
#    Consumers verify this header as described in RFC 6455 §10 conventions.
#    Key is the shared webhook secret; body is the canonical JSON encoding.
# ---------------------------------------------------------------------------

def sign_payload(payload: dict, secret: str) -> str:
    """Return hex digest of HMAC-SHA256 over the canonical JSON body."""
    # Compact encoding ensures the signature is stable regardless of
    # the client's JSON serializer key ordering.
    body = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode()
    return hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()


# ---------------------------------------------------------------------------
# 3. Async dispatcher with exponential backoff + full jitter
#    The ClientSession is created once per worker in the lifespan handler
#    above — never one per request, and never one per retry attempt.
# ---------------------------------------------------------------------------

async def deliver_with_retry(
    url: str,
    payload: dict,
    secret: str,
    max_retries: int = 4,
) -> bool:
    """
    Attempt delivery up to max_retries times.
    Returns True on success, False after exhausting retries (caller routes to DLQ).

    Backoff formula: min(cap, 2^attempt) + uniform(0, 1)
    Cap at 30 s to avoid very long waits on the last attempt.
    """
    headers = {
        "Content-Type": "application/json",
        "X-Webhook-Signature": f"sha256={sign_payload(payload, secret)}",
        "X-Event-ID": payload["event_id"],
        "X-Feature-ID": payload["feature_id"],
    }
    timeout = aiohttp.ClientTimeout(total=10)

    for attempt in range(max_retries):
        try:
            assert _session is not None, "ClientSession not initialised"
            async with _session.post(
                url, json=payload, headers=headers, timeout=timeout
            ) as resp:
                if resp.status < 300:
                    log.info(
                        "delivered event=%s to=%s attempt=%d",
                        payload["event_id"], url, attempt + 1,
                    )
                    return True

                # 4xx = client-side error; retrying will not help
                if 400 <= resp.status < 500:
                    log.error(
                        "client error status=%d event=%s url=%s — routing to DLQ",
                        resp.status, payload["event_id"], url,
                    )
                    return False

                # 5xx = server-side transient; fall through to retry
                log.warning(
                    "server error status=%d event=%s attempt=%d",
                    resp.status, payload["event_id"], attempt + 1,
                )

        except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
            log.warning(
                "network error event=%s attempt=%d error=%s",
                payload["event_id"], attempt + 1, exc,
            )

        if attempt < max_retries - 1:
            # Full jitter: actual delay is uniform in [0, capped_base]
            base = min(30.0, 2 ** attempt)
            delay = random.uniform(0, base)
            await asyncio.sleep(delay)

    log.error(
        "exhausted retries event=%s url=%s — routing to DLQ",
        payload["event_id"], url,
    )
    return False


async def route_to_dlq(payload: dict, reason: str) -> None:
    """
    In production: write to a Redis LPUSH / Kafka topic / database row.
    Preserve event_id, feature_id, original payload, failure reason, and
    timestamp so the event can be replayed or audited without data loss.
    """
    record = {
        "dlq_timestamp": datetime.now(timezone.utc).isoformat(),
        "reason": reason,
        "payload": payload,
    }
    # Replace this log statement with your broker write:
    log.error("DLQ record: %s", json.dumps(record))


# ---------------------------------------------------------------------------
# 4. FastAPI endpoint
#    In production, replace the direct POST with a broker consumer loop that
#    reads from Redis Streams / Kafka topics and calls deliver_with_retry.
# ---------------------------------------------------------------------------

@app.post("/dispatch")
async def dispatch_event(
    event: GeoEvent,
    x_target_endpoint: str = Header(..., description="Consumer endpoint URL"),
    x_webhook_secret: str = Header(default="change-me"),
) -> dict:
    """
    Validate and dispatch a single geospatial event.

    Headers
    -------
    X-Target-Endpoint : full URL of the consumer webhook endpoint
    X-Webhook-Secret  : shared HMAC secret for payload signing
    """
    payload = event.model_dump(mode="json")
    delivered = await deliver_with_retry(x_target_endpoint, payload, x_webhook_secret)

    if not delivered:
        await route_to_dlq(payload, reason="delivery_failed")
        raise HTTPException(
            status_code=502,
            detail=f"Could not deliver event {event.event_id}; routed to DLQ",
        )

    return {"status": "delivered", "event_id": event.event_id}

Parameter / option reference

Parameter Type Spatial constraint Default
geometry dict RFC 7946 GeoJSON geometry; coordinates in WGS 84 (EPSG:4326); no self-intersections required
feature_id str Any opaque string; used as broker ordering key required
change_type "create" | "update" | "delete" — required
properties dict Arbitrary key-value pairs; avoid embedding full geometry duplicates here {}
max_retries int — 4
timeout (per attempt) float (seconds) Keep ≤ 15 s; large polygon serialization can inflate response time 10
X-Webhook-Secret str Minimum 32 random bytes in production; rotate on breach "change-me"

Gotchas and spatial edge cases

  1. Coordinate ring orientation. RFC 7946 requires exterior rings to be counter-clockwise and interior rings (holes) to be clockwise. Shapely’s is_valid check does not enforce RFC 7946 winding order — it only checks topological validity. Call shapely.geometry.mapping(geom) after geom = orient(geom, sign=1.0) from shapely.ops before serializing to JSON.

  2. CRS mismatch on ingestion. PostGIS may emit geometries in a projected CRS such as EPSG:3857 (Web Mercator). Passing those coordinates directly into a GeoJSON payload violates RFC 7946. Reproject to WGS 84 (EPSG:4326) using pyproj.Transformer before the Pydantic validator runs, as described in Handling Mixed CRS Payloads in Python Event Handlers.

  3. Precision loss during serialization. Python’s json.dumps serializes floats with up to 17 significant digits, but model_dump(mode="json") from Pydantic may round coordinates via intermediate float conversion. For high-precision geometries (cadastral surveys, survey-grade GPS), serialize coordinates explicitly to a fixed decimal place (e.g., 7 d.p. ≈ 1 cm accuracy) using a custom JSON encoder rather than relying on default float formatting.

  4. Empty geometry after transformation. Reprojection of very small or degenerate geometries can produce an empty Shapely result. Always call geom.is_empty after any transformation and reject the event with a descriptive error rather than forwarding an empty geometry downstream.

  5. HMAC signature drift on large payloads. If the consumer reconstructs the body from a different key ordering than the dispatcher’s sort_keys=True, the HMAC will not match. Standardize on sort_keys=True, separators=(",", ":") on both sides, and document this contract in your API reference. Mismatches surface as 401s, not 5xxs, so they bypass retry logic — log them explicitly.

  6. Session lifecycle in async frameworks. Creating a new aiohttp.ClientSession inside deliver_with_retry (one per attempt) opens a new TCP connection and TLS handshake on every retry. The code above creates one session per worker process via the lifespan context manager — the deprecated @app.on_event("startup") hook still works but is being phased out, so prefer lifespan on new services. When using Celery workers instead of FastAPI, create the session in a celery.signals.worker_process_init handler and close it in worker_process_shutdown.

Minimal verification snippet

Run this with pytest against a live instance or a respx-mocked session to confirm end-to-end correctness, including signature verification.

python 
# test_dispatcher.py
import asyncio
import hashlib
import hmac
import json

import pytest
from httpx import AsyncClient, ASGITransport

from geospatial_webhook_dispatcher import app, sign_payload

VALID_POINT_GEOMETRY = {
    "type": "Point",
    "coordinates": [-122.4194, 37.7749],   # San Francisco, WGS 84 (EPSG:4326)
}

VALID_POLYGON_GEOMETRY = {
    "type": "Polygon",
    "coordinates": [[
        [-122.42, 37.78],
        [-122.40, 37.78],
        [-122.40, 37.76],
        [-122.42, 37.76],
        [-122.42, 37.78],   # closed ring
    ]],
}


@pytest.mark.asyncio
async def test_sign_payload_is_deterministic():
    payload = {"event_id": "abc", "feature_id": "f1", "geometry": VALID_POINT_GEOMETRY}
    sig1 = sign_payload(payload, "secret")
    sig2 = sign_payload(payload, "secret")
    assert sig1 == sig2, "Signature must be deterministic"


@pytest.mark.asyncio
async def test_sign_payload_hmac_correct():
    payload = {"event_id": "abc", "feature_id": "f1"}
    expected_body = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode()
    expected = hmac.new(b"secret", expected_body, hashlib.sha256).hexdigest()
    assert sign_payload(payload, "secret") == expected


@pytest.mark.asyncio
async def test_dispatch_rejects_invalid_geometry():
    self_intersecting = {
        "type": "Polygon",
        "coordinates": [[
            [0, 0], [2, 2], [2, 0], [0, 2], [0, 0],  # bowtie — self-intersecting
        ]],
    }
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        resp = await client.post(
            "/dispatch",
            json={
                "feature_id": "f1",
                "geometry": self_intersecting,
                "change_type": "update",
            },
            headers={
                "X-Target-Endpoint": "http://example.com/hook",
                "X-Webhook-Secret": "test-secret",
            },
        )
    assert resp.status_code == 422, "Self-intersecting polygon must be rejected at validation"


@pytest.mark.asyncio
async def test_dispatch_rejects_missing_geometry_type():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        resp = await client.post(
            "/dispatch",
            json={
                "feature_id": "f2",
                "geometry": {"coordinates": [0, 0]},  # missing "type"
                "change_type": "create",
            },
            headers={
                "X-Target-Endpoint": "http://example.com/hook",
                "X-Webhook-Secret": "test-secret",
            },
        )
    assert resp.status_code == 422

Frequently asked questions

Why use async dispatch instead of synchronous HTTP for geospatial webhooks?

Spatial payloads — especially polygon and multipolygon geometries — can be large, and topology validation is CPU-bound. Blocking on synchronous requests.post calls ties up workers and pushes backpressure into the ingestion layer, so a slow consumer slows down the database trigger that produced the event. Async delivery with aiohttp decouples validation from delivery: the ingestion path stays write-optimized while a fixed pool of connections fans out POSTs concurrently.

How do I handle CRS mismatches before dispatching a webhook payload?

Normalize every incoming geometry to WGS 84 (EPSG:4326) at the validation layer using pyproj.Transformer before serializing to GeoJSON, since RFC 7946 mandates EPSG:4326. Reject or reproject any geometry whose source CRS differs, and log the original SRID so consumers can audit the transformation. The full reprojection workflow is covered in Handling Mixed CRS Payloads in Python Event Handlers.

When should events go to a dead-letter queue versus be retried?

Retry on transient failures — 5xx responses, timeouts, and connection resets — using the capped exponential backoff with full jitter shown above. Send straight to the dead-letter queue on 4xx client errors (bad endpoint, auth failure, malformed signature), because retrying those only wastes connections, and after max_retries is exhausted for transient errors. Always preserve the full original payload plus failure metadata in the DLQ so the event can be replayed without data loss.