Parsing GeoJSON Webhooks with FastAPI and Pydantic

Define RFC 7946-compliant Pydantic v2 models with extra="forbid", verify the HMAC signature over the raw request body, then call model_validate_json so FastAPI rejects malformed GeoJSON with a structured 422 before any spatial logic runs. This how-to sits under GeoJSON to Protobuf Mapping, part of Spatial Payload Routing & Parsing.

When to use this pattern

Schema enforcement at the HTTP edge is worth the extra model code when malformed geometry would otherwise propagate silently:

  • You ingest GeoJSON from third parties or untrusted clients — partner integrations, mobile clients, or IoT gateways — where a single missing type discriminator or a swapped longitude/latitude pair can corrupt a spatial index downstream.
  • You need deterministic, machine-readable rejection: a 422 with field-level error paths beats a 500 traceback when a sender debugs why their payload bounced, and it keeps malformed coordinates out of Geometry Validation Pipelines further along.
  • Your endpoint feeds a binary serializer or async fan-out — for example, converting validated features as covered in GeoJSON to Protobuf Mapping — where the consumer assumes the input is already structurally sound.

If you are only logging payloads or your producer is fully trusted and inside your own trust boundary, a permissive dict[str, Any] parse may be enough. Reserve strict modeling for ingress points where “is this even valid GeoJSON?” is a question you cannot afford to answer downstream.

Validation flow

The diagram below shows the fail-fast order. Cheap checks run first (signature over raw bytes, then structural parse); the spatial bound and ring checks run inside the model only once the JSON is well-formed. Any failure short-circuits to a typed HTTP error before the payload reaches your processor.

GeoJSON webhook validation flow A flowchart showing four sequential gates: read raw body and verify HMAC, parse JSON into Pydantic models, validate coordinate bounds and ring closure, then accept and route downstream. Signature failures return 401, structural failures return 422, and accepted payloads return 202 to an async processor. Incoming POST 1. Verify HMAC over raw bytes 2. Parse JSON into Pydantic models 3. Spatial checks bounds · ring closure valid GeoJSON ✓ 4. Accept event 202 → async processor Reject before processing 401 bad signature · 422 invalid GeoJSON malformed out of bounds

Complete runnable code block

The handler below is self-contained. It models the GeoJSON object graph with discriminated unions, enforces WGS 84 / EPSG:4326 coordinate bounds and closed polygon rings inside the model, verifies an HMAC-SHA256 signature over the raw bytes, and returns 202 with a fast hand-off to an async processor. Because the spec mandates EPSG:4326 with longitude before latitude, the bound checks assume that order.

python 
import hashlib
import hmac
from typing import Annotated, Any, Literal, Union

from fastapi import FastAPI, Header, HTTPException, Request, status
from pydantic import BaseModel, ConfigDict, Field, ValidationError, model_validator

app = FastAPI(title="GeoJSON Webhook Ingestor")
WEBHOOK_SECRET = b"load-from-secrets-manager-not-source-control"

# --- Coordinate helpers (RFC 7946 mandates EPSG:4326, longitude first) ---
def _check_position(pos: list[float]) -> None:
    # A position is [lon, lat] or [lon, lat, elevation]; only the first two are bounded.
    if len(pos) < 2:
        raise ValueError("position must have at least longitude and latitude")
    lon, lat = pos[0], pos[1]
    if not (-180.0 <= lon <= 180.0):
        raise ValueError(f"longitude {lon} outside [-180, 180]")
    if not (-90.0 <= lat <= 90.0):
        raise ValueError(f"latitude {lat} outside [-90, 90]")

def _walk_positions(coords: Any) -> None:
    # Recurse into the nested coordinate array until we hit [lon, lat, ...] positions.
    if coords and isinstance(coords[0], (int, float)):
        _check_position(coords)
    else:
        for child in coords:
            _walk_positions(child)

# --- Pydantic v2 GeoJSON models ---
class Point(BaseModel):
    model_config = ConfigDict(extra="forbid")
    type: Literal["Point"]
    coordinates: list[float]

class Polygon(BaseModel):
    model_config = ConfigDict(extra="forbid")
    type: Literal["Polygon"]
    # Polygon = array of linear rings; each ring is an array of positions.
    coordinates: list[list[list[float]]]

    @model_validator(mode="after")
    def rings_must_close(self) -> "Polygon":
        for ring in self.coordinates:
            if len(ring) < 4:
                raise ValueError("a linear ring needs at least 4 positions")
            if ring[0] != ring[-1]:
                raise ValueError("polygon ring is not closed (first != last position)")
        return self

# Discriminated union keeps error messages precise per geometry type.
Geometry = Annotated[Union[Point, Polygon], Field(discriminator="type")]

class Feature(BaseModel):
    model_config = ConfigDict(extra="forbid")
    type: Literal["Feature"]
    geometry: Geometry
    properties: dict[str, Any] | None = None
    id: str | int | None = None

    @model_validator(mode="after")
    def coordinates_in_bounds(self) -> "Feature":
        _walk_positions(self.geometry.coordinates)
        return self

class FeatureCollection(BaseModel):
    model_config = ConfigDict(extra="forbid")
    type: Literal["FeatureCollection"]
    features: list[Feature] = Field(min_length=1)

class WebhookPayload(BaseModel):
    model_config = ConfigDict(extra="forbid")
    event_type: str
    data: Feature | FeatureCollection

# --- Security ---
def verify_signature(raw: bytes, signature: str) -> bool:
    expected = hmac.new(WEBHOOK_SECRET, raw, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature)

# --- Route ---
@app.post("/webhooks/geojson", status_code=status.HTTP_202_ACCEPTED)
async def ingest(
    request: Request,
    x_webhook_signature: str = Header(..., alias="X-Webhook-Signature"),
):
    # 1. Read raw bytes ONCE; the signature is computed over these exact bytes.
    raw = await request.body()
    if not verify_signature(raw, x_webhook_signature):
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid signature")

    # 2. Parse the same bytes into the typed model; structured 422 on any violation.
    try:
        payload = WebhookPayload.model_validate_json(raw)
    except ValidationError as exc:
        raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, exc.errors())

    # 3. Acknowledge fast; offload heavy spatial work to a queue or executor.
    await route_downstream(payload)
    return {"status": "accepted", "event_type": payload.event_type}

async def route_downstream(payload: WebhookPayload) -> None:
    # Hand off to PostGIS, a broker, or a protobuf serializer here.
    ...

Parameter / option reference table

The settings below are the ones that change correctness, not just style. Spatial constraints assume RFC 7946 (EPSG:4326, longitude first).

Option Type Spatial constraint / effect Default
ConfigDict(extra="forbid") bool flag Rejects unknown keys (e.g. a typo’d geometery) instead of silently dropping them extra="ignore"
Field(discriminator="type") union discriminator Routes each geometry to the right model so errors name the actual type, not “none matched” none
Field(min_length=1) on features int A FeatureCollection with zero features is structurally legal in RFC 7946 but usually a producer bug unbounded
model_validate_json(raw) classmethod Parses bytes directly; skips a json.loads round-trip and preserves byte-exact signature input n/a
hmac.compare_digest function Constant-time comparison; prevents timing side-channels that leak the secret use == (unsafe)
lon bound float range -180.0 <= lon <= 180.0; values outside indicate swapped lat/lon or a non-EPSG:4326 source none
lat bound float range -90.0 <= lat <= 90.0; a value of 200 is a classic swapped-axis symptom none
ring closure list check First and last position of each linear ring must be equal not enforced

Gotchas and spatial edge cases

  1. Swapped longitude and latitude. The single most common bug. RFC 7946 mandates [longitude, latitude], but many producers (and most people) say “lat, lon”. A Point at [40.7, -74.0] will pass naive parsing yet place New York in the Indian Ocean. The bound checks above catch only egregious swaps (lat > 90); for plausible mid-latitude values you need a sanity check against an expected region.
  2. Coordinate precision loss before signing. If you round coordinates before computing or verifying the HMAC, the recomputed digest will not match. Sign the raw bytes exactly as received and round only after validation, never before.
  3. Ring orientation is not validity. RFC 7946 prefers right-hand-rule winding (exterior counter-clockwise, holes clockwise) but most parsers accept either. Pydantic checks closure, not winding. If a downstream consumer assumes a winding order, normalize it explicitly with Shapely’s orient() after parsing.
  4. extra="ignore" hides typos. Without extra="forbid", a payload sending geometery (misspelled) parses with geometry=None and no error. Forbidding extras turns that into a loud 422.
  5. Unbounded coordinate arrays are a denial-of-service surface. A single Polygon with millions of vertices will exhaust memory during parse. Cap request body size at the gateway and consider a vertex-count guard before handing the geometry to Async Processing for Heavy Geometries.
  6. CRS assumptions on merge. This handler assumes EPSG:4326. If any producer emits EPSG:3857 or a local grid, the bound checks will reject legitimate data or — worse — pass garbage. Normalize upstream per CRS Normalization Strategies before this endpoint, or carry an explicit CRS field and transform inside the validator.

Minimal verification snippet

Run the following with pytest. It signs a valid payload, asserts a 202, then mutates the body and asserts the typed rejections — no live server required, thanks to FastAPI’s TestClient.

python 
import hashlib
import hmac
import json

from fastapi.testclient import TestClient

from app import WEBHOOK_SECRET, app  # import from the module above

client = TestClient(app)

def _sign(body: bytes) -> dict[str, str]:
    sig = hmac.new(WEBHOOK_SECRET, body, hashlib.sha256).hexdigest()
    return {"X-Webhook-Signature": sig}

VALID = {
    "event_type": "feature.created",
    "data": {
        "type": "Feature",
        "geometry": {"type": "Point", "coordinates": [-74.006, 40.7128]},
        "properties": {"name": "NYC"},
    },
}

def test_valid_payload_accepted():
    body = json.dumps(VALID).encode()
    resp = client.post("/webhooks/geojson", content=body, headers=_sign(body))
    assert resp.status_code == 202
    assert resp.json()["event_type"] == "feature.created"

def test_out_of_bounds_latitude_rejected():
    bad = json.loads(json.dumps(VALID))
    bad["data"]["geometry"]["coordinates"] = [-74.006, 200.0]  # lat > 90
    body = json.dumps(bad).encode()
    resp = client.post("/webhooks/geojson", content=body, headers=_sign(body))
    assert resp.status_code == 422

def test_bad_signature_rejected():
    body = json.dumps(VALID).encode()
    resp = client.post(
        "/webhooks/geojson", content=body,
        headers={"X-Webhook-Signature": "deadbeef"},
    )
    assert resp.status_code == 401

A passing run confirms three guarantees at once: well-formed GeoJSON is accepted, an out-of-range coordinate is rejected with 422 before any processing, and a forged signature never reaches the parser.