Schema Validation for Spatial Records

Schema validation for spatial records is the foundational control plane for any OGC-compliant data publishing pipeline. When geographic datasets move from ingestion to catalog publication, structural integrity, coordinate reference system (CRS) compliance, and metadata alignment must all be verified before indexing. Without deterministic validation, malformed geometries, missing mandatory fields, or non-conforming XML and JSON payloads corrupt downstream services, break spatial queries, and trigger compliance failures in government and enterprise environments.

This guide is part of the Spatial Metadata & Catalog Integration section. It provides a production-ready validation workflow for GeoJSON, OGC API Features, and ISO 19139 spatial records, with tested Python patterns, step-by-step execution logic, and error-routing strategies for platform engineers.

---

Pipeline Architecture Overview

Validation is not a single gate — it is a sequence of five deterministic stages. Each stage isolates one class of failure so that downstream systems receive only verified payloads and failed records carry explicit error codes for remediation.

---

Prerequisites & Architecture Context

This guide assumes Python 3.10+, familiarity with HTTP and JSON Schema, and a working understanding of what a GeoJSON FeatureCollection and an ISO 19139 gmd:MD_Metadata document look like. Before implementing validation pipelines, ensure your environment meets the following baseline:

• Core libraries: pydantic>=2.0, jsonschema, shapely>=2.0, pyproj, lxml, defusedxml
• Schema references: RFC 7946 (GeoJSON), the official JSON Schema specification, ISO 19115 XSD files (stored locally — never fetched at runtime), and DCAT-AP SHACL shapes
• CRS registry: pyproj’s bundled EPSG database for resolving URNs such as urn:ogc:def:crs:EPSG::4326
• Observability: structured JSON logging and a dead-letter queue (DLQ) for failed payloads, provisioned before deploying validation logic to production

The SRS and Coordinate Reference System Handling guide covers axis-order rules and on-the-fly reprojection in detail — reviewing it first significantly reduces CRS-related integration friction. For the broader catalog publishing context this pipeline feeds into, see Automated Metadata Harvesting Workflows.

---

Specification Deep-Dive

GeoJSON & OGC API Features Validation Rules

RFC 7946 defines a precise set of structural and geometric constraints for GeoJSON. Many spatial pipelines treat GeoJSON as permissive JSON, which leads to silent downstream failures. The mandatory structural elements are:

OGC API – Features Part 1 (Core) extends GeoJSON with a conformance layer. Response collections must include a links array with self, alternate, and next relations. Feature responses must declare a numberMatched and numberReturned at the collection level. These are mandatory for harvesting compliance but are absent from plain GeoJSON schema validation — check them separately.

ISO 19139 XSD Validation Complexity

ISO 19139 (the XML encoding of ISO 19115 metadata standards) involves a chain of interdependent XSD files across multiple namespaces: gmd, gco, gml, xsi, and optionally srv for service metadata. Validation failures almost always trace to one of these root causes:

JSON Schema Fragment for GeoJSON FeatureCollection

The following minimal JSON Schema fragment enforces the mandatory structural rules for a GeoJSON FeatureCollection:

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "type": "object",
  "required": ["type", "features"],
  "properties": {
    "type": { "const": "FeatureCollection" },
    "features": {
      "type": "array",
      "items": {
        "type": "object",
        "required": ["type", "geometry", "properties"],
        "properties": {
          "type": { "const": "Feature" },
          "geometry": {
            "oneOf": [
              { "type": "null" },
              {
                "type": "object",
                "required": ["type", "coordinates"],
                "properties": {
                  "type": {
                    "enum": ["Point","MultiPoint","LineString",
                             "MultiLineString","Polygon","MultiPolygon"]
                  }
                }
              }
            ]
          },
          "properties": { "type": ["object", "null"] }
        }
      }
    }
  }
}

---

Python Implementation

The following implementation demonstrates a type-safe, multi-stage validation function. It separates structural, geometric, and CRS checks while accumulating all errors before raising — enabling batch remediation without multiple round-trips.

"""
spatial_validator.py — Multi-stage spatial record validator.
Requires: pydantic>=2.0, jsonschema, shapely>=2.0, pyproj, lxml, defusedxml (Python 3.10+)
"""

from __future__ import annotations

import json
import logging
from dataclasses import dataclass, field
from typing import Any

import defusedxml.ElementTree as safe_ET
from jsonschema import Draft202012Validator, ValidationError as JsonSchemaError
from lxml import etree
from pyproj import CRS, Transformer
from shapely.geometry import shape
from shapely.validation import make_valid

logger = logging.getLogger(__name__)

# ── Error container ────────────────────────────────────────────────────────────

class SpatialValidationError(Exception):
    """Raised when one or more validation stages produce errors."""

    def __init__(self, errors: list[str]) -> None:
        self.errors = errors
        super().__init__(f"Validation failed ({len(errors)} error(s)): " + "; ".join(errors))

# ── Stage 1: Ingest & structural JSON Schema validation ────────────────────────

def validate_geojson_structure(
    payload: dict[str, Any],
    schema: dict[str, Any],
) -> list[str]:
    """
    Validate payload against a JSON Schema.
    Returns a list of error messages; empty list means structurally valid.
    """
    validator = Draft202012Validator(schema)
    return [
        f"Schema violation at {'/'.join(str(p) for p in e.absolute_path) or 'root'}: {e.message}"
        for e in validator.iter_errors(payload)
    ]

# ── Stage 2: Geometry & topology verification ──────────────────────────────────

def validate_geometries(
    features: list[dict[str, Any]],
    auto_repair: bool = False,
) -> list[str]:
    """
    Validate geometry objects in a GeoJSON feature list.
    When auto_repair=True, attempts make_valid on invalid rings before failing.
    """
    errors: list[str] = []

    for idx, feat in enumerate(features):
        ref = feat.get("id", f"index:{idx}")
        geom = feat.get("geometry")

        if geom is None:
            # Explicit null geometry is valid per RFC 7946; absent key is not
            if "geometry" not in feat:
                errors.append(f"Feature {ref}: 'geometry' key absent (must be null or an object)")
            continue

        try:
            shp = shape(geom)
        except Exception as exc:
            errors.append(f"Feature {ref}: geometry parse error — {exc}")
            continue

        if not shp.is_valid:
            if auto_repair:
                repaired = make_valid(shp)
                if repaired.is_valid:
                    logger.warning(
                        "Feature %s: geometry auto-repaired (was: %s)", ref, shp.wkt[:120]
                    )
                    shp = repaired
                else:
                    errors.append(f"Feature {ref}: irreparable invalid geometry — {shp.wkt[:80]}")
                    continue
            else:
                errors.append(f"Feature {ref}: invalid geometry — {shp.wkt[:80]}")
                continue

        minx, miny, maxx, maxy = shp.bounds
        if not (-180.0 <= minx <= maxx <= 180.0 and -90.0 <= miny <= maxy <= 90.0):
            errors.append(
                f"Feature {ref}: coordinates out of geographic bounds "
                f"(bounds: {minx:.4f},{miny:.4f},{maxx:.4f},{maxy:.4f})"
            )

    return errors

# ── Stage 3: CRS resolution & axis-order verification ─────────────────────────

def validate_crs(
    declared_crs: str | None,
    target_epsg: int = 4326,
) -> list[str]:
    """
    Resolve the declared CRS string using pyproj and verify it is
    transformable to target_epsg. Returns a list of error messages.
    """
    if not declared_crs:
        # GeoJSON defaults to CRS84 (lon/lat WGS84) when absent — treat as valid
        return []

    errors: list[str] = []
    try:
        src_crs = CRS.from_user_input(declared_crs)
    except Exception as exc:
        return [f"CRS resolution failed for '{declared_crs}': {exc}"]

    if src_crs.is_deprecated:
        errors.append(
            f"CRS '{declared_crs}' is deprecated; replace with its successor "
            f"({src_crs.name})"
        )

    try:
        Transformer.from_crs(src_crs, CRS.from_epsg(target_epsg), always_xy=True)
    except Exception as exc:
        errors.append(f"CRS '{declared_crs}' cannot be transformed to EPSG:{target_epsg}: {exc}")

    return errors

# ── Orchestrator ───────────────────────────────────────────────────────────────

@dataclass
class SpatialRecordValidator:
    """
    Orchestrates all validation stages for a GeoJSON FeatureCollection payload.
    Instantiate once; call validate() per record for best performance.
    """
    geojson_schema: dict[str, Any]
    target_epsg: int = 4326
    auto_repair_geometries: bool = False
    # Accumulated errors from the last validate() call
    last_errors: list[str] = field(default_factory=list, init=False, repr=False)

    def validate(self, payload: dict[str, Any]) -> dict[str, Any]:
        """
        Run all validation stages.
        Returns the (possibly geometry-repaired) payload on success.
        Raises SpatialValidationError on any failure.
        """
        errors: list[str] = []

        # Stage 1 — structural schema check
        errors.extend(validate_geojson_structure(payload, self.geojson_schema))

        # Stage 2 — geometry (run even if schema errors exist, for full error set)
        features = payload.get("features", [])
        errors.extend(
            validate_geometries(features, auto_repair=self.auto_repair_geometries)
        )

        # Stage 3 — CRS (legacy GeoJSON CRS member; OGC API Features uses Link headers)
        declared = (payload.get("crs") or {}).get("properties", {}).get("name")
        errors.extend(validate_crs(declared, target_epsg=self.target_epsg))

        self.last_errors = errors
        if errors:
            raise SpatialValidationError(errors)

        return payload

Annotated Walkthrough

validate_geojson_structure uses Draft202012Validator.iter_errors rather than the validate() shortcut so every schema violation is collected before raising. The absolute_path attribute on each error points to the exact JSON Pointer location of the violation, making error messages immediately actionable.

validate_geometries calls shape(geom) from Shapely to parse the raw GeoJSON geometry dict into a geometry object. Shapely accepts topologically invalid geometries during construction — only shp.is_valid reveals topology errors. When auto_repair=True, make_valid rebuilds the geometry using the JTS-compatible algorithm (e.g. self-intersecting polygons are split into valid sub-polygons). Always log the original WKT before repair; audit requirements in government environments mandate this.

validate_crs calls CRS.from_user_input which accepts EPSG codes, URN strings (urn:ogc:def:crs:EPSG::4326), WKT strings, and PROJ strings. Checking src_crs.is_deprecated catches legacy codes such as EPSG:4001 before they reach the catalog indexer. The Transformer.from_crs call verifies the transformation path exists without mutating coordinates — actual reprojection happens downstream during indexing.

---

Error Handling & Edge Cases

Axis-order ambiguity between EPSG:4326 and CRS84. EPSG:4326 defines axes as latitude first, longitude second. CRS84 — the default CRS for GeoJSON per RFC 7946 — is longitude first, latitude second. Always pass always_xy=True to pyproj.Transformer to force XY (longitude, latitude) order regardless of the EPSG axis definition. Failing to do this silently swaps coordinates during reprojection. The SRS and Coordinate Reference System Handling guide covers this trap in detail.

ISO 19139 XXE injection. Always parse untrusted ISO 19139 XML with defusedxml or lxml with resolve_entities=False. Never pass raw catalog responses directly to etree.fromstring — some legacy CSW endpoints embed external entity references that trigger server-side request forgery when resolved.

Empty geometry vs absent geometry. RFC 7946 §3.2 permits "geometry": null as a valid Feature (features without location). An absent geometry key is a structural violation. Validate for key presence separately from null-check, as many spatial ETL tools silently drop the key rather than setting it to null.

Polygon ring orientation. RFC 7946 §3.1.6 recommends exterior rings be counter-clockwise and interior rings (holes) clockwise, but this is not a strict requirement. Some catalog backends (PostGIS, Elasticsearch geo_shape) enforce orientation internally and will silently reorder or reject rings. Normalise ring orientation as part of validation to avoid non-deterministic indexing behaviour.

Feature ID collisions in batch ingestion. When validating FeatureCollections sourced from multiple upstream systems, feature.id values are not guaranteed unique across sources. Add a namespace prefix derived from the source endpoint URL before passing records to the catalog upsert layer.

---

XML Validation for ISO 19139 Records

For ISO 19139 payloads harvested via CSW or OGC API – Records, XSD-based validation replaces JSON Schema. The following pattern uses lxml’s XMLSchema validator with locally cached schema files:

from lxml import etree
from pathlib import Path

def validate_iso19139(
    raw_xml: str | bytes,
    xsd_root: Path,
) -> list[str]:
    """
    Validate an ISO 19139 XML document against local XSD files.
    xsd_root must contain gmd/gmd.xsd (and its imports: gco, gml, xlink).
    Returns a list of validation error messages; empty list means valid.
    """
    # Build a resolver that maps public schema URIs to local files
    class LocalResolver(etree.Resolver):
        def resolve(self, url: str, id: str, context):  # type: ignore[override]
            name = url.rstrip("/").split("/")[-1]
            local = xsd_root / name
            if local.exists():
                return self.resolve_filename(str(local), context)
            return None

    parser = etree.XMLParser(resolve_entities=False)
    parser.resolvers.add(LocalResolver())

    xsd_path = xsd_root / "gmd" / "gmd.xsd"
    try:
        schema_doc = etree.parse(str(xsd_path), parser)
        schema = etree.XMLSchema(schema_doc)
    except Exception as exc:
        return [f"XSD load error: {exc}"]

    try:
        if isinstance(raw_xml, str):
            raw_xml = raw_xml.encode("utf-8")
        doc = etree.fromstring(raw_xml, parser)
    except etree.XMLSyntaxError as exc:
        return [f"XML parse error: {exc}"]

    schema.validate(doc)
    return [str(e) for e in schema.error_log]

Note that etree.XMLSchema.error_log accumulates all violations rather than raising on the first — mirror this behaviour in your orchestrator so the entire error set is emitted to the dead-letter queue in one pass.

---

Testing & Compliance Verification

A minimal unit test skeleton covering the three validation stages:

import json
import unittest

MINIMAL_SCHEMA = {
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "type": "object",
    "required": ["type", "features"],
    "properties": {
        "type": {"const": "FeatureCollection"},
        "features": {"type": "array"},
    },
}

VALID_PAYLOAD: dict = {
    "type": "FeatureCollection",
    "features": [
        {
            "type": "Feature",
            "geometry": {"type": "Point", "coordinates": [13.405, 52.520]},
            "properties": {"name": "Berlin"},
        }
    ],
}

class TestSpatialValidator(unittest.TestCase):

    def setUp(self) -> None:
        self.validator = SpatialRecordValidator(geojson_schema=MINIMAL_SCHEMA)

    def test_valid_payload_passes(self) -> None:
        result = self.validator.validate(VALID_PAYLOAD)
        self.assertEqual(result["type"], "FeatureCollection")

    def test_missing_features_key_fails(self) -> None:
        bad = {"type": "FeatureCollection"}
        with self.assertRaises(SpatialValidationError) as ctx:
            self.validator.validate(bad)
        self.assertTrue(any("features" in e for e in ctx.exception.errors))

    def test_out_of_bounds_coordinates_fail(self) -> None:
        bad_coords = dict(VALID_PAYLOAD)
        bad_coords["features"] = [
            {
                "type": "Feature",
                "geometry": {"type": "Point", "coordinates": [200.0, 52.520]},
                "properties": {},
            }
        ]
        with self.assertRaises(SpatialValidationError) as ctx:
            self.validator.validate(bad_coords)
        self.assertTrue(any("bounds" in e for e in ctx.exception.errors))

    def test_invalid_crs_urn_fails(self) -> None:
        errors = validate_crs("urn:ogc:def:crs:EPSG::99999")
        self.assertTrue(len(errors) > 0)

    def test_valid_crs_urn_passes(self) -> None:
        errors = validate_crs("urn:ogc:def:crs:EPSG::4326")
        self.assertEqual(errors, [])

For OGC CITE compliance testing, the OGC Validation System provides a hosted TEAM Engine instance that tests WFS and OGC API – Features endpoints against official conformance classes. Run your service through the Features Core and GeoJSON conformance suites before promoting to production. The OGC Standards Architecture & Service Fundamentals overview explains the conformance class hierarchy.

---

Error Routing & Retry Architecture

Validation failures should not block the entire ingestion pipeline. Implement a two-tier routing strategy based on the error’s remediation path:

Hard failures are records that cannot be promoted regardless of downstream effort: structural schema violations, unparseable payloads, or missing mandatory metadata. Route these directly to a dead-letter queue (DLQ) with status: rejected. Store the full error list and a SHA-256 hash of the original payload for audit traceability. These require upstream source correction or manual intervention.

Soft failures have a defined repair path: topology issues that make_valid can address, deprecated CRS codes that map to a current equivalent, or missing optional fields that default values can supply. Route these to a retry queue with an attached remediation script and exponential backoff. Log each repair action with the before/after state.

Emit structured JSON log entries for every validation decision:

import hashlib
import json
import logging

def emit_validation_log(
    record_id: str,
    payload_bytes: bytes,
    errors: list[str],
    stage: str,
    status: str,  # "passed" | "rejected" | "retry"
) -> None:
    logger.info(json.dumps({
        "record_id": record_id,
        "validation_stage": stage,
        "status": status,
        "error_count": len(errors),
        "errors": errors,
        "payload_hash": hashlib.sha256(payload_bytes).hexdigest(),
    }))

Government and enterprise environments often require immutable audit trails. Store validation reports alongside catalog entries using content-addressable storage keyed on payload_hash. This satisfies data-lineage requirements and enables reproducible re-validation when schemas are updated.

---

Performance & Scaling Notes

Compile schemas once. Draft202012Validator(schema) and etree.XMLSchema(schema_doc) parse and compile the schema on instantiation. Construct these objects once at application startup — not per record — and reuse them across the request lifetime. On high-throughput pipelines, schema compilation is a significant CPU cost if repeated naively.

Batch geometry validation. For large FeatureCollections (tens of thousands of features), consider validating geometry in batches using Shapely’s vectorised operations via geopandas.GeoDataFrame.is_valid before falling back to per-feature Shapely object construction. This reduces Python loop overhead significantly.

CRS resolution caching. CRS.from_user_input performs a registry lookup on each call. Cache resolved CRS objects in a module-level dict keyed by the declared CRS string, and cache Transformer objects keyed by (src_crs_auth_code, target_epsg) tuples. This is especially important when processing FeatureCollections where all features share the same CRS.

Parallelising across endpoints. Use concurrent.futures.ProcessPoolExecutor (not threads) for CPU-bound geometry validation. Shapely releases the GIL for some operations but not all. Each worker process should construct its own SpatialRecordValidator instance with pre-compiled schemas loaded from a shared file path.

lxml memory management. When validating thousands of ISO 19139 XML documents, lxml element trees accumulate in memory. Call root.clear() and then del root after extracting the data you need. For streaming validation of large catalog responses, use etree.iterparse with the events=("end",) pattern to process and discard elements incrementally.

---

Gotchas & Frequently Asked Questions

Should I fail-fast or accumulate all validation errors before rejecting a record?

Accumulate all errors within each validation stage before raising. This lets you route the complete error set to the dead-letter queue in one pass and avoids multiple round-trips for batch remediation. Reserve fail-fast only for the very first structural check — if the payload cannot be parsed at all, further stages cannot run. Within structural, geometry, and CRS stages, always collect the full set.

When is it safe to auto-repair invalid geometry with make_valid?

Auto-repair is appropriate for known precision-loss artifacts: slightly self-intersecting rings caused by coordinate truncation during reprojection, and duplicate vertices introduced by simplification. It is not appropriate when the root cause is incorrect digitising or a corrupted source file — repaired geometry may be geometrically valid but semantically wrong. Always log the original WKT alongside the repaired result, and flag auto-repaired records for downstream human review before final catalog promotion.

How do I resolve the axis-order ambiguity between EPSG:4326 and CRS84?

EPSG:4326 defines axes as latitude first, longitude second. CRS84 — the implicit CRS for GeoJSON and OGC API – Features — is longitude first, latitude second. Pass always_xy=True to pyproj.Transformer.from_crs to force XY (longitude, latitude) order regardless of the EPSG axis definition. Validate incoming coordinates against both conventions when the source CRS declaration is ambiguous or absent, and log the assumed axis order in the validation report.

How do I validate ISO 19139 XML without fetching XSDs over the network?

Download the gmd, gco, gml, and xlink schema files from schemas.opengis.net and store them under a local directory tree that mirrors the original path structure. Implement a custom etree.Resolver subclass (as shown in the XML validation section above) that intercepts resolve() calls and returns local file paths. This eliminates network latency, firewall issues, and external availability risk from your validation pipeline.

How do I classify a failure as hard (dead-letter) vs. soft (retry)?

Hard failures have no programmatic repair path: missing mandatory metadata fields, unparseable encoding, coordinates that fall completely outside geographic bounds, or CRS strings that cannot be resolved at all. Soft failures have a defined repair path: topology errors that make_valid addresses, deprecated CRS codes that map to a known successor, or missing optional fields that default values can supply. Classify by whether your own code can deterministically produce a valid record from the failed one. If not, it is a hard failure.

---

Back to Spatial Metadata & Catalog Integration

Related

• Automated Metadata Harvesting Workflows — how harvested records flow into this validation pipeline, with dead-letter queue and incremental sync patterns
• Implementing ISO 19115 Metadata Standards — element structure and mandatory field mapping for the ISO 19139 records this page validates
• DCAT-AP for Spatial Data Portals — downstream serialisation step once records pass validation and are ready for cross-jurisdictional portal publication
• SRS and Coordinate Reference System Handling — axis-order rules, EPSG URN resolution, and on-the-fly reprojection used in the CRS validation stage
• OGC Standards Architecture & Service Fundamentals — OGC service contract model and conformance class hierarchy underpinning GeoJSON and OGC API – Features validation