Geospatial Schema Standardization & Mapping Automation

Enforcing Attribute Domains with pandera Schemas

Most teams end up enforcing attribute rules three times: once in the ETL job, once in a test fixture, and once in whatever script a data steward runs before accepting a delivery. Three implementations of one rule set drift within weeks, and the drift is invisible — each of them passes its own tests. This procedure compiles the single declarative manifest described in attribute domain and code list validation into a pandera schema object, and then uses that one object everywhere. It belongs to Gate 4 of Spatial Data Quality Validation & Geometry Integrity.

The steps map to configure (Step 1, compilation), execute (Step 2, lazy validation), validate (Step 3, converting failures into report rows), and log (Step 4, the same schema in CI and pre-commit). Nothing here overrides the section’s rule that domain validation never repairs a value — pandera’s coercion features are deliberately left switched off.

Prerequisites checklist

Step 1: Compile the manifest into a pandera schema

python
# quality/pandera_compile.py — pandera >=0.18, pandas >=2.1 — Python 3.10+
import pandera as pa
from pandera import Check, Column, DataFrameSchema

from quality.registers import load_pinned

DTYPES = {"identifier": str, "enumeration": str, "register": str,
          "range_numeric": float, "range_date": "datetime64[ns]"}


def column_for(spec: dict) -> Column:
    kind = spec["kind"]
    checks: list[Check] = []

    if kind == "enumeration":
        values = spec["values"]
        checks.append(Check.isin(values, name="ATTR_UNDEFINED_CODE",
                                 error=f"not one of {values}"))
    elif kind == "register":
        active, _retired, _parents = load_pinned(spec)
        checks.append(Check.isin(sorted(active), name="ATTR_UNDEFINED_CODE",
                                 error=f"not an active member of {spec['register_uri']}"))
    elif kind == "identifier":
        if spec.get("pattern"):
            checks.append(Check.str_matches(spec["pattern"], name="ATTR_PATTERN_MISMATCH"))
    elif kind == "range":
        if spec.get("minimum") is not None:
            checks.append(Check.ge(spec["minimum"], name="ATTR_OUT_OF_RANGE"))
        if spec.get("maximum") is not None:
            checks.append(Check.le(spec["maximum"], name="ATTR_OUT_OF_RANGE"))

    return Column(
        dtype=DTYPES.get(f"range_{spec.get('unit', 'numeric')}", str) if kind == "range" else str,
        checks=checks,
        nullable=not spec["required"],
        unique=spec.get("unique", False),
        coerce=False,                 # never repair: a cast here would hide upstream drift
        required=True,                # the COLUMN must exist even when values may be null
        name=spec["name"],
    )


def compile_schema(manifest: dict) -> DataFrameSchema:
    return DataFrameSchema(
        columns={f["name"]: column_for(f) for f in manifest["fields"]},
        strict=False,                 # extra columns are the mapping stage's business
        ordered=False,
        name=manifest["dataset"],
    )

Two settings carry real weight. coerce=False is the line that keeps this stage honest: with coercion on, a string "12" in a numeric column becomes 12 and the fact that the source sent text is lost, which is precisely the drift the type coercion stage is supposed to surface. And the distinction between required=True on the column and nullable on its values matters: a missing column is a schema failure that stops the run, while a null value in an optional field is normal data.

Step 2: Validate lazily so one run reports every failure

python
# quality/validate.py — pandera >=0.18 — Python 3.10+
import pandera as pa


def validate(df, schema) -> pa.errors.SchemaErrors | None:
    try:
        schema.validate(df, lazy=True)       # lazy: collect all failures, do not stop at the first
        return None
    except pa.errors.SchemaErrors as errors:
        return errors

lazy=True is not a convenience. Without it, a delivery with four separate problems takes four full pipeline runs to diagnose, each one revealing the next failure — and each run over a multi-gigabyte delivery. With it, one run produces a complete failure-case frame, and the steward gets a single, actionable list.

Step 3: Convert failure cases into rejection rows

pandera returns failures as a frame with the check name, the failing value and the row index. Mapping it onto the quality report is mechanical, and the mapping is where the check names chosen in Step 1 pay off.

python
# quality/validate.py — pandas >=2.1 — Python 3.10+
def to_rejections(errors: pa.errors.SchemaErrors, df, run_id: str) -> list[dict]:
    cases = errors.failure_cases          # columns: schema_context, column, check, failure_case, index
    rows = []
    for record in cases.to_dict("records"):
        index = record["index"]
        rows.append({
            "feature_id": df.at[index, "feature_id"] if index is not None else "",
            "run_id":     run_id,
            "field":      record["column"],
            "code":       record["check"] or "ATTR_CHECK_FAILED",   # the name set in Step 1
            "observed":   str(record["failure_case"])[:120],
            "expected":   record["schema_context"],
        })
    return rows

Naming each Check with the quality report’s code — ATTR_UNDEFINED_CODE, ATTR_OUT_OF_RANGE — means no translation table is needed here, and the codes in the published report are the same strings a developer reads in the schema definition. Keeping observed is what makes the report actionable rather than a count; truncating it at 120 characters keeps a pathological value from bloating the store.

One Manifest, One Compiled Schema, Three Consumers A single manifest file feeds a compile step that produces one pandera DataFrameSchema object. Three consumers read that object: the ETL job validates each delivery and writes rejection rows, the test suite validates fixture data including a negative control, and a pre-commit hook validates any fixture committed to the repository. Because all three read the same compiled object, a rule change reaches every consumer at once and none of them can drift. attribute_domains.yaml the only place rules are written compile_schema() one DataFrameSchema object ETL job writes rejection rows per delivery test suite fixtures plus a negative control pre-commit hook any fixture entering the repository compile

Step 4: Reuse the same schema in tests and pre-commit

python
# tests/test_attribute_domains.py — pytest >=7, pandera >=0.18
import pandas as pd
import pytest

from quality.pandera_compile import compile_schema
from quality.manifest import load_domain_manifest

SCHEMA = compile_schema(load_domain_manifest("attribute_domains.yaml"))


def test_valid_delivery_passes():
    df = pd.read_parquet("tests/fixtures/parcels_clean.parquet")
    SCHEMA.validate(df, lazy=True)          # raises on any failure


def test_every_defect_class_is_caught():
    df = pd.read_parquet("tests/fixtures/parcels_with_domain_defects.parquet")
    with pytest.raises(pa.errors.SchemaErrors) as excinfo:
        SCHEMA.validate(df, lazy=True)
    codes = set(excinfo.value.failure_cases["check"])
    assert {"ATTR_UNDEFINED_CODE", "ATTR_OUT_OF_RANGE", "ATTR_PATTERN_MISMATCH"} <= codes


def test_coercion_is_disabled():
    # Negative control: a numeric column receiving text must FAIL, not be quietly cast.
    df = pd.DataFrame({"feature_id": ["a"], "building_height_m": ["12"], "land_use": ["1_1_1"],
                       "ownership_status": ["public"], "parcel_id": ["14-023-9"]})
    with pytest.raises(pa.errors.SchemaErrors):
        SCHEMA.validate(df, lazy=True)
yaml
# .pre-commit-config.yaml
repos:
  - repo: local
    hooks:
      - id: validate-fixtures
        name: validate committed fixtures against the domain manifest
        entry: python -m quality.validate_fixtures
        language: system
        files: '^tests/fixtures/.*\.(parquet|gpkg)$'

The pre-commit hook prevents the quiet failure that undermines every rule set eventually: a fixture edited to make a failing test pass. If a fixture must violate the manifest — the defect corpus does, by design — it lives in a directory the hook excludes, and that exclusion is itself visible in the configuration.

Verification

A clean delivery produces no output and a non-zero exit only on failure. A delivery with problems produces the complete list in one pass:

text
SchemaErrors: 3 schema error(s) found in dataframe
  ATTR_UNDEFINED_CODE   land_use            failure_case='residental'   (1,412 rows)
  ATTR_OUT_OF_RANGE     building_height_m   failure_case=984.0          (3 rows)
  ATTR_PATTERN_MISMATCH parcel_id           failure_case='14-23-9'      (1 row)

The row counts are the triage signal: 1,412 identical failures is one source-system misconfiguration, three scattered ones are individual records, and the single pattern mismatch is almost certainly a leading zero lost to a spreadsheet.

Troubleshooting

Symptom Likely cause Fix
Validation reports only one failure lazy=True omitted Always validate lazily in batch contexts; eager validation is only for interactive debugging
A numeric column reports every value as failing Column arrived as object dtype because a single row held text Fix at the casting stage; do not enable coerce to make it pass
failure_cases has null indices The failure is at schema level — a missing column — not at row level Handle schema-level failures separately; they stop the run rather than quarantine features
Register-backed check is slow on large frames Check.isin against a large list rebuilt per call Compile the schema once per run, not per partition; load_pinned should be cached
Rules pass in CI but fail in the ETL job Two schemas compiled from two manifest copies There must be exactly one manifest path; make it a package resource rather than a relative path