Validating WKT and GeoJSON with Pydantic v2

← Back to Strict Pydantic Validation for Geometry

Use Pydantic v2’s BeforeValidator with Shapely to parse, repair, and bounds-check WKT and GeoJSON geometry at request time — before any PostGIS write occurs.

# Context & When to Use

FastAPI routes that accept geometry payloads face two formats in practice: Well-Known Text (WKT), which GIS desktop tools and GDAL pipelines export natively, and GeoJSON, the format web clients and mapping SDKs send. Both formats carry identical topology risks — self-intersecting rings, flipped coordinate order, coordinates outside valid geographic bounds — that PostGIS will silently accept or raise cryptic errors for depending on the operation.

Placing validation in a Pydantic BeforeValidator is the right approach when you need to catch these errors at the HTTP boundary and surface them as structured 422 Unprocessable Entity responses, rather than letting malformed geometry reach the database layer and cause partial writes or confusing ST_IsValid failures. This pattern complements the broader Strict Pydantic Validation for Geometry strategy documented in the parent guide.

Prefer this technique over a plain @field_validator when you need to normalize the input (e.g., re-serialize WKT with consistent rounding) before Pydantic’s strict type checking evaluates it. The BeforeValidator runs first; strict mode runs after. If you instead used mode="after", Pydantic would attempt type coercion on the raw string before your logic runs, and strict=True would reject most inputs before you could parse them.

Preconditions: Python 3.10+, pydantic>=2.5, shapely>=2.0, and fastapi>=0.100. The make_valid() function used below requires Shapely 2.x — it is not available in Shapely 1.x.

# Validation Pipeline

The diagram below shows how a request body moves through the validation layers before reaching the database:

# Runnable Implementation

The implementation below provides two reusable Annotated type aliases — WKTGeometry and GeoJSONGeometry — that you drop into any Pydantic model. A model_validator(mode="before") rejects cross-type mismatches cheaply before the expensive Shapely parse runs, and a model_validator(mode="after") enforces EPSG:4326 coordinate bounds on the final parsed geometry.

from __future__ import annotations

import json
from typing import Annotated, Any, Literal

from pydantic import BaseModel, BeforeValidator, Field, model_validator
from shapely import wkt
from shapely.geometry import mapping, shape
from shapely.validation import make_valid


# ---------------------------------------------------------------------------
# Parse functions — called by BeforeValidator before strict typing is applied
# ---------------------------------------------------------------------------

def parse_wkt(value: Any) -> str:
    """Accept a WKT string, parse it with Shapely, repair if needed,
    then re-serialize with consistent 6-decimal-place rounding."""
    if not isinstance(value, str):
        raise ValueError("WKT geometry must be a plain string")
    try:
        geom = wkt.loads(value)
        if not geom.is_valid:
            geom = make_valid(geom)  # repairs self-intersections & ring issues
        return wkt.dumps(geom, rounding_precision=6)
    except Exception as exc:
        raise ValueError(f"Invalid WKT: {exc}") from exc


def parse_geojson(value: Any) -> dict:
    """Accept a GeoJSON geometry as a dict or raw JSON string.
    Parse, repair, then return a normalized dict via shapely.mapping()."""
    if isinstance(value, str):
        try:
            value = json.loads(value)
        except json.JSONDecodeError as exc:
            raise ValueError(f"Malformed GeoJSON string: {exc}") from exc
    if not isinstance(value, dict):
        raise ValueError("GeoJSON geometry must be a dict or valid JSON string")
    try:
        geom = shape(value)          # validates structure + coordinate order
        if not geom.is_valid:
            geom = make_valid(geom)
        return mapping(geom)         # returns a clean, normalized dict
    except Exception as exc:
        raise ValueError(f"Invalid GeoJSON geometry: {exc}") from exc


# ---------------------------------------------------------------------------
# Annotated type aliases — reuse these across all your Pydantic models
# ---------------------------------------------------------------------------

WKTGeometry = Annotated[str, BeforeValidator(parse_wkt)]
GeoJSONGeometry = Annotated[dict, BeforeValidator(parse_geojson)]


# ---------------------------------------------------------------------------
# Payload model with pre-check and post-bounds validation
# ---------------------------------------------------------------------------

class SpatialPayload(BaseModel):
    model_config = {"strict": True, "extra": "forbid"}

    geometry_type: Literal["wkt", "geojson"]
    # Union resolves left-to-right; the pre-check below picks the right branch.
    geometry: WKTGeometry | GeoJSONGeometry
    precision: int = Field(default=6, ge=0, le=12)
    srid: int = Field(default=4326, description="EPSG code; defaults to WGS84")

    @model_validator(mode="before")
    @classmethod
    def enforce_type_match(cls, data: Any) -> Any:
        """Reject mismatches before Shapely parses the value.
        Avoids running expensive geometry parsing on obviously wrong input."""
        if isinstance(data, dict):
            gtype = data.get("geometry_type")
            geom = data.get("geometry")
            if gtype == "wkt" and not isinstance(geom, str):
                raise ValueError("Expected a string for geometry_type='wkt'")
            if gtype == "geojson" and not isinstance(geom, (dict, str)):
                raise ValueError("Expected dict/JSON string for geometry_type='geojson'")
        return data

    @model_validator(mode="after")
    def validate_4326_bounds(self) -> "SpatialPayload":
        """For EPSG:4326 GeoJSON payloads, verify every coordinate pair lies
        within [-180,180] longitude and [-90,90] latitude (RFC 7946)."""
        if self.srid != 4326 or not isinstance(self.geometry, dict):
            return self

        flat_coords: list[float] = []

        def _flatten(node: Any) -> None:
            if isinstance(node, (int, float)):
                flat_coords.append(float(node))
            elif isinstance(node, list):
                for item in node:
                    _flatten(item)

        _flatten(self.geometry.get("coordinates", []))

        # GeoJSON coordinate order: [longitude, latitude, optional_elevation]
        for i in range(0, len(flat_coords) - 1, 2):
            lon, lat = flat_coords[i], flat_coords[i + 1]
            if not (-180 <= lon <= 180):
                raise ValueError(
                    f"Longitude {lon} is outside EPSG:4326 bounds [-180, 180]"
                )
            if not (-90 <= lat <= 90):
                raise ValueError(
                    f"Latitude {lat} is outside EPSG:4326 bounds [-90, 90]"
                )
        return self


# ---------------------------------------------------------------------------
# FastAPI route — validation errors surface automatically as 422 responses
# ---------------------------------------------------------------------------

from fastapi import FastAPI

app = FastAPI()


@app.post("/geometries/validate", status_code=200)
async def validate_geometry(payload: SpatialPayload) -> dict:
    """Validate and normalize an inbound WKT or GeoJSON geometry.
    Returns the cleaned geometry and its detected type.

    On validation failure FastAPI converts pydantic.ValidationError
    into a structured 422 Unprocessable Entity with a 'detail' array."""
    return {
        "geometry_type": payload.geometry_type,
        "geometry": payload.geometry,
        "srid": payload.srid,
        "precision": payload.precision,
    }

For endpoints that accept bulk uploads — multiple geometries in one request — see the Handling Async File Uploads for Shapefile Processing guide, which builds on this same validator pattern for Celery-backed ingestion pipelines.

# Key Parameters & Options

# Gotchas & Failure Modes

• make_valid() changes the geometry type. A self-intersecting Polygon may become a GeometryCollection or MultiPolygon. If downstream code assumes a specific WKB type for the PostGIS column (GEOMETRY(Polygon, 4326)), the insert will fail with ERROR: Geometry type (MultiPolygon) does not match column type (Polygon). Check geom.geom_type after repair and reject unexpected type changes, or widen the column definition to GEOMETRY(Geometry, 4326).

• WKT from GDAL omits the SRID prefix. GDAL exports plain WKT like POLYGON ((…)), not SRID=4326;POLYGON ((…)). If you pass EWKT (Extended WKT with the SRID= prefix) to shapely.wkt.loads(), it raises shapely.errors.WKTReadingError. Strip the prefix before parsing: value.split(";", 1)[-1].

• GeoJSON coordinates in the wrong order. The GeoJSON spec (RFC 7946) mandates [longitude, latitude]. Many sources — especially older WFS feeds — emit [latitude, longitude]. Shapely will parse either without error, but your bounds-check validator catches the swap only if latitude exceeds ±180. Add an explicit swap-detection step if you consume third-party feeds.

• strict=True blocks JSON numbers coerced to Python int/float. If srid arrives as "4326" (a string) rather than 4326 (an integer), Pydantic raises Input should be a valid integer under strict mode. Document the contract clearly in your OpenAPI schema and test with real clients.

• Very large MultiPolygon payloads spike memory. shapely.geometry.shape() loads the entire coordinate array into Python objects. A 100 MB GeoJSON response body can exceed 500 MB of resident memory during parsing. Enforce a Content-Length limit in your ASGI middleware (app.add_middleware(ContentSizeLimitMiddleware, max_content_size=5_000_000)) before the request reaches the validator.

# Verification Snippet

Start the application with uvicorn app:app --reload, then test with curl:

# Valid GeoJSON polygon — should return 200 with normalized geometry
curl -s -X POST http://localhost:8000/geometries/validate \
  -H "Content-Type: application/json" \
  -d '{
    "geometry_type": "geojson",
    "geometry": {
      "type": "Polygon",
      "coordinates": [[[0,0],[1,0],[1,1],[0,1],[0,0]]]
    },
    "srid": 4326
  }' | python3 -m json.tool

# Malformed WKT — should return 422 with structured error detail
curl -s -X POST http://localhost:8000/geometries/validate \
  -H "Content-Type: application/json" \
  -d '{"geometry_type": "wkt", "geometry": "NOT VALID WKT", "srid": 4326}' \
  | python3 -m json.tool

# Coordinate out of bounds — longitude 200 exceeds [-180,180]
curl -s -X POST http://localhost:8000/geometries/validate \
  -H "Content-Type: application/json" \
  -d '{
    "geometry_type": "geojson",
    "geometry": {
      "type": "Point",
      "coordinates": [200, 45]
    },
    "srid": 4326
  }' | python3 -m json.tool

Expected 422 response shape for the malformed WKT request:

{
  "detail": [
    {
      "type": "value_error",
      "loc": ["body", "geometry"],
      "msg": "Value error, Invalid WKT: …",
      "input": "NOT VALID WKT"
    }
  ]
}

For a minimal unit test without spinning up FastAPI:

import pytest
from pydantic import ValidationError
from app import SpatialPayload  # adjust import to your module path

def test_valid_geojson_point():
    payload = SpatialPayload(
        geometry_type="geojson",
        geometry={"type": "Point", "coordinates": [13.405, 52.52]},
        srid=4326,
    )
    assert payload.geometry["type"] == "Point"

def test_rejects_out_of_bounds_longitude():
    with pytest.raises(ValidationError, match="Longitude 200"):
        SpatialPayload(
            geometry_type="geojson",
            geometry={"type": "Point", "coordinates": [200, 52]},
            srid=4326,
        )

def test_rejects_type_mismatch():
    with pytest.raises(ValidationError, match="Expected a string"):
        SpatialPayload(
            geometry_type="wkt",
            geometry={"type": "Point", "coordinates": [0, 0]},
            srid=4326,
        )

# Related

• Strict Pydantic Validation for Geometry — parent guide covering ring orientation, topology checks, and CRS alignment strategies
• Implementing ST_Within and ST_Intersects in FastAPI — how validated geometry flows into PostGIS spatial filter queries
• GeoJSON vs GeoParquet Serialization — choosing the right serialization format for your API responses

← Back to Strict Pydantic Validation for Geometry