Handling Mixed CRS Payloads in Python Webhooks

import logging
from functools import lru_cache
from typing import Optional, Dict, Any

from fastapi import FastAPI
from pydantic import BaseModel, Field
from pyproj import Transformer, CRS
from pyproj.exceptions import CRSError
from shapely.geometry import shape, mapping
from shapely.validation import make_valid
from shapely.ops import transform as shp_transform

app = FastAPI()
logger = logging.getLogger("crs_normalizer")

TARGET_CRS = "EPSG:4326"  # RFC 7946 canonical output for GeoJSON


class SpatialPayload(BaseModel):
    geometry: Dict[str, Any]              # GeoJSON-like geometry object
    crs: Optional[str] = None            # explicit CRS, e.g. "EPSG:3857"
    metadata: Dict[str, Any] = Field(default_factory=dict)


@lru_cache(maxsize=256)
def get_transformer(source_crs: str, target_crs: str) -> Transformer:
    # Compiling a PROJ pipeline is expensive; cache by the (source, target)
    # EPSG pair so repeated payloads from the same CRS reuse one object.
    # always_xy=True forces (x=longitude, y=latitude) order — PROJ 6+ otherwise
    # honours the authority axis order and silently swaps lon/lat for EPSG:4326.
    return Transformer.from_crs(
        CRS.from_user_input(source_crs),
        CRS.from_user_input(target_crs),
        always_xy=True,
    )


def resolve_source_crs(payload: SpatialPayload) -> str:
    # 1. Explicit field wins.
    if payload.crs:
        return payload.crs
    # 2. GeoJSON 2008-style crs member (deprecated by RFC 7946 but still emitted).
    member = payload.geometry.get("crs")
    if isinstance(member, dict):
        name = member.get("properties", {}).get("name")
        if name:
            return name
    # 3. Inference: geographic longitude/latitude never exceed |180|/|90|.
    #    A coordinate magnitude in the thousands means a projected grid, so a
    #    missing CRS that looks projected must NOT be assumed to be WGS84.
    x, y = _first_coordinate(payload.geometry)
    if abs(x) > 180 or abs(y) > 90:
        raise ValueError(
            f"coordinates ({x}, {y}) look projected but no CRS was supplied"
        )
    # 4. Per RFC 7946, GeoJSON with no CRS is WGS84.
    return TARGET_CRS


def _first_coordinate(geometry: Dict[str, Any]) -> tuple[float, float]:
    coords = geometry.get("coordinates")
    while isinstance(coords, list) and coords and isinstance(coords[0], list):
        coords = coords[0]
    if not isinstance(coords, list) or len(coords) < 2:
        raise ValueError("geometry has no usable coordinates")
    return float(coords[0]), float(coords[1])


def normalize(payload: SpatialPayload) -> Dict[str, Any]:
    source_crs = resolve_source_crs(payload)
    try:
        transformer = get_transformer(source_crs, TARGET_CRS)
    except CRSError as exc:
        raise ValueError(f"unresolvable CRS '{source_crs}': {exc}") from exc

    geom = shape(payload.geometry)
    geom = shp_transform(transformer.transform, geom)

    # A datum shift can introduce or expose self-intersections; re-check here,
    # not before the transform, so validity reflects the OUTPUT geometry.
    if not geom.is_valid:
        geom = make_valid(geom)
        if not geom.is_valid:
            raise ValueError("geometry still invalid after make_valid")

    return {
        "geometry": mapping(geom),
        "crs": TARGET_CRS,
        "source_crs": source_crs,        # keep provenance for auditing
        "metadata": payload.metadata,
    }


@app.post("/ingest")
async def ingest(payload: SpatialPayload):
    try:
        normalized = normalize(payload)
    except ValueError as exc:
        logger.warning("quarantining payload: %s", exc)
        # Push to a dead-letter / quarantine topic with the original input.
        return {"status": "quarantined", "reason": str(exc),
                "original": payload.model_dump()}
    # Publish `normalized` to your broker (Kafka, SQS, Redis Streams, ...).
    return {"status": "normalized", "payload": normalized}

import math
from your_handler import normalize, SpatialPayload  # adjust import path


def test_web_mercator_point_normalizes_to_wgs84():
    # EPSG:3857 metres for roughly (lon=0, lat=0) — the origin.
    payload = SpatialPayload(
        geometry={"type": "Point", "coordinates": [0.0, 0.0]},
        crs="EPSG:3857",
    )
    out = normalize(payload)
    lon, lat = out["geometry"]["coordinates"]
    assert out["crs"] == "EPSG:4326"
    assert math.isclose(lon, 0.0, abs_tol=1e-6)
    assert math.isclose(lat, 0.0, abs_tol=1e-6)
    assert out["source_crs"] == "EPSG:3857"


def test_unlabelled_projected_coords_are_quarantined():
    payload = SpatialPayload(
        geometry={"type": "Point", "coordinates": [512345.0, 4781002.0]},
        crs=None,  # no CRS, magnitudes far exceed |180|/|90|
    )
    try:
        normalize(payload)
    except ValueError as exc:
        assert "projected" in str(exc)
    else:
        raise AssertionError("expected ValueError for unlabelled projected input")