SpatiaLite & GeoPackage Automation with Python

How to Write a Data Quality Gate for a Field Dataset

Run the checks against the temporary artefact, before the atomic rename that publishes it. Structural first because it is cheapest, then the domain rules,…

Run the checks against the temporary artefact, before the atomic rename that publishes it. Structural first because it is cheapest, then the domain rules, then the comparison against the previous run — and exit non-zero on any blocking failure so the rename never happens and consumers keep yesterday’s good data.

This page belongs to the Data Quality Gates & Monitoring guide, which sets out the three kinds of check this script runs.

Why This Matters

A check that runs after publication is a report. It tells you that the data consumers are already using is wrong, which is useful and is not what anyone wanted. The same check, moved ten lines earlier in the script, is a gate: the bad artefact never becomes visible, and the failure costs a re-run rather than a recall.

The position relative to the rename is the whole difference. Everything else here — which checks, in which order, with which severities — is refinement.

Prerequisites

  • A pipeline that builds into a temporary path and publishes by rename
  • sqlite3 on the path, and Python 3.9+ for the comparison step
  • A manifest from the previous successful run
  • The severity bands decided in advance, per the parent guide

Primary Method

bash
#!/usr/bin/env bash
# gate.sh — run every band, then publish only if all of them passed
set -euo pipefail

TMP=${1:?usage: gate.sh <built.gpkg> <destination.gpkg> <previous.manifest.json>}
DEST=${2:?}
PREV=${3:-}

fail() { printf 'GATE FAIL: %s\n' "$*" >&2; exit 1; }

# --- band 1: structural -------------------------------------------------
sqlite3 "$TMP" "PRAGMA quick_check;" | grep -q '^ok$' \
    || fail "quick_check reported corruption"

python3 conformance.py "$TMP" || fail "conformance checks failed"

# --- band 2: absolute domain rules --------------------------------------
python3 domain_rules.py "$TMP" || fail "domain rules violated"

# --- band 3: relative, against the previous run -------------------------
python3 manifest.py "$TMP" > "${TMP%.gpkg}.manifest.json"
if [ -n "$PREV" ] && [ -f "$PREV" ]; then
    python3 compare_manifests.py "${TMP%.gpkg}.manifest.json" "$PREV" \
        || fail "drift against the previous run"
else
    printf 'GATE NOTE: no previous manifest — baseline established\n'
fi

# --- publish ------------------------------------------------------------
sqlite3 "$TMP" "PRAGMA wal_checkpoint(TRUNCATE); PRAGMA journal_mode=DELETE;"
mv -f "$TMP" "$DEST"
cp -f "${TMP%.gpkg}.manifest.json" "${DEST%.gpkg}.manifest.json"
printf 'published %s\n' "$DEST"

set -e plus the fail helper is what makes this a gate rather than a sequence of checks: any non-zero exit aborts before the mv, and the previously published artefact stays exactly where it was.

Band order, and where the publish sitsThe structural band runs first because it is cheapest and because a corrupt container makes every later check meaningless. The absolute domain rules run second. The relative comparison against the previous run runs third, because it needs the manifest the run has just produced. Only after all three pass does the checkpoint and atomic rename publish the artefact. A failure at any band exits before the rename.1 · structuralquick_check + conformance · cheapest, and everything else assumes it2 · absolutedomain rules · returns the offending rows, not a boolean3 · relativecompared with the previous manifest · catches plausible nonsensepublishcheckpoint, then rename · the first moment a consumer can see anything
A failure in any of the three upper bands exits before the green one, which is the entire mechanism.

Step-by-Step Walkthrough

1. Write the domain rules as queries returning rows

python
# domain_rules.py — each rule returns the rows that violate it
import sqlite3
import sys

RULES = [
    ("parcel_ref is missing",
     "SELECT fid FROM parcels WHERE parcel_ref IS NULL OR trim(parcel_ref) = ''"),

    ("surveyed_on is in the future",
     "SELECT fid FROM observations WHERE surveyed_on > date('now')"),

    ("geometry outside the survey extent",
     "SELECT fid FROM parcels "
     "WHERE ST_MinX(geom) < -8.2 OR ST_MaxX(geom) > 1.8 "
     "   OR ST_MinY(geom) < 49.9 OR ST_MaxY(geom) > 60.9"),

    ("duplicate parcel_ref",
     "SELECT parcel_ref FROM parcels GROUP BY parcel_ref HAVING count(*) > 1"),
]


def main(path: str) -> int:
    conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
    conn.enable_load_extension(True)
    conn.load_extension("mod_spatialite")

    failures = 0
    for name, sql in RULES:
        rows = conn.execute(sql).fetchall()
        if rows:
            failures += 1
            print(f"FAIL {name}: {len(rows)} row(s), e.g. {rows[:5]}", file=sys.stderr)
    conn.close()
    return 1 if failures else 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1]))

The extent rule is the highest-value line in that file. It costs almost nothing — ST_MinX reads the geometry header rather than decoding coordinates — and it catches an entire family of projection and datum errors, because a mangled coordinate almost always lands far outside the area the survey covers.

2. Emit a manifest the next run can compare against

python
# manifest.py — describe what this run produced
import json, sqlite3, sys
from pathlib import Path


def describe(path: str) -> dict:
    conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
    layers = [r[0] for r in conn.execute(
        "SELECT table_name FROM gpkg_contents WHERE data_type = 'features'"
    )]
    out = {
        "container": Path(path).name,
        "bytes": Path(path).stat().st_size,
        "layers": {},
    }
    for layer in sorted(layers):
        n = conn.execute(f'SELECT count(*) FROM "{layer}"').fetchone()[0]
        cols = [(r[1], r[2]) for r in conn.execute(f'PRAGMA table_info("{layer}")')]
        out["layers"][layer] = {"rows": n, "columns": cols}
    conn.close()
    return out


if __name__ == "__main__":
    print(json.dumps(describe(sys.argv[1]), indent=2))

3. Compare against the previous run

python
# compare_manifests.py — the band that catches plausible nonsense
import json, sys
from pathlib import Path

TOLERANCE = 0.10          # a layer may move 10% between runs without comment


def compare(now: dict, was: dict) -> list[str]:
    problems = []

    for gone in sorted(set(was["layers"]) - set(now["layers"])):
        problems.append(f"layer disappeared: {gone}")
    for added in sorted(set(now["layers"]) - set(was["layers"])):
        problems.append(f"layer appeared: {added}")

    for name, before in was["layers"].items():
        after = now["layers"].get(name)
        if after is None or before["rows"] == 0:
            continue
        drift = (after["rows"] - before["rows"]) / before["rows"]
        if abs(drift) > TOLERANCE:
            problems.append(
                f"{name}: {before['rows']} -> {after['rows']} ({drift:+.1%})"
            )
        if after["columns"] != before["columns"]:
            problems.append(f"{name}: schema changed")

    return problems


if __name__ == "__main__":
    issues = compare(
        json.loads(Path(sys.argv[1]).read_text()),
        json.loads(Path(sys.argv[2]).read_text()),
    )
    for line in issues:
        print("DRIFT:", line, file=sys.stderr)
    sys.exit(1 if issues else 0)
Setting the drift tolerance from observed behaviourA tolerance set from a single run has no basis and will either fire constantly or never. A tolerance set from two weeks of observed run-to-run movement reflects how the data actually behaves. A layer that grows steadily wants a tighter band than one that is genuinely bursty, so per-layer tolerances are worth the extra configuration once the baseline exists.guessed from one runno basis in the datafires constantly, or nevermeasured over two weeksreflects real variabilityfires when something changedsteady layertight band, a few per centany jump is meaningfulbursty layerwider band, or a rolling mediana single previous run misleads
Collect manifests before enforcing. Two weeks of observation costs nothing and is what stops the gate being switched off.

4. Provide an override that records a reason

A gate with no override gets commented out the first time it blocks a legitimate large change. One with an override that demands a reason stays in place and leaves an audit trail.

bash
# gate.sh, near the top
OVERRIDE_REASON=${GATE_OVERRIDE:-}

# ...and in the relative band
if ! python3 compare_manifests.py "$NEW_MANIFEST" "$PREV"; then
    if [ -z "$OVERRIDE_REASON" ]; then
        fail "drift against the previous run (set GATE_OVERRIDE=<reason> to proceed)"
    fi
    printf 'GATE OVERRIDE: %s\n' "$OVERRIDE_REASON"
    python3 - "$NEW_MANIFEST" "$OVERRIDE_REASON" <<'PY'
import json, sys
from pathlib import Path
p = Path(sys.argv[1]); m = json.loads(p.read_text())
m["override_reason"] = sys.argv[2]
p.write_text(json.dumps(m, indent=2))
PY
fi

Writing the reason into the manifest means the override travels with the artefact, so a question six months later about why the count halved has an answer in the file rather than in someone’s memory.

5. Make the failure output usable

bash
fail() {
    printf '\n' >&2
    printf 'GATE FAIL: %s\n' "$*" >&2
    printf '  artefact: %s (not published)\n' "$TMP" >&2
    printf '  previous: %s (still live)\n' "$DEST" >&2
    printf '  manifest: %s\n' "${TMP%.gpkg}.manifest.json" >&2
    exit 1
}

Naming the un-published artefact matters: the person investigating needs to open it, and a message that says only “gate failed” sends them looking for it.

Verification

The gate must be shown to refuse. Break a copy three ways and confirm each band catches its own.

bash
#!/usr/bin/env bash
# test-gate.sh — prove each band rejects
set -uo pipefail

expect_fail() {
    if ./gate.sh "$1" /tmp/out.gpkg prev.manifest.json >/dev/null 2>&1; then
        echo "GATE FAILED TO STOP: $2"; exit 1
    fi
    echo "ok — rejected: $2"
}

cp good.gpkg /tmp/t1.gpkg
sqlite3 /tmp/t1.gpkg "DROP TABLE gpkg_geometry_columns;"
expect_fail /tmp/t1.gpkg "dropped registry table"

cp good.gpkg /tmp/t2.gpkg
sqlite3 /tmp/t2.gpkg "UPDATE observations SET surveyed_on='2099-01-01' WHERE fid=1;"
expect_fail /tmp/t2.gpkg "future date"

cp good.gpkg /tmp/t3.gpkg
sqlite3 /tmp/t3.gpkg "DELETE FROM parcels WHERE fid > 100;"
expect_fail /tmp/t3.gpkg "most of a layer deleted"

Also confirm the good path publishes and that a failure leaves the destination untouched:

bash
cp good.gpkg /tmp/ok.gpkg
./gate.sh /tmp/ok.gpkg /srv/out.gpkg prev.manifest.json
test -f /srv/out.gpkg || { echo "good artefact was not published"; exit 1; }

BEFORE=$(sha256sum /srv/out.gpkg | cut -d' ' -f1)
cp good.gpkg /tmp/bad.gpkg
sqlite3 /tmp/bad.gpkg "DELETE FROM parcels;"
./gate.sh /tmp/bad.gpkg /srv/out.gpkg prev.manifest.json || true
AFTER=$(sha256sum /srv/out.gpkg | cut -d' ' -f1)
[ "$BEFORE" = "$AFTER" ] || { echo "a failed run modified the published artefact"; exit 1; }
Three deliberate corruptions, one per bandDropping a registry table must be caught by the structural band. Setting a survey date in the future must be caught by the absolute band. Deleting most of a layer must be caught by the relative band. Each corruption targets exactly one band, so a test that passes proves that band works rather than that some other band happened to notice.drop a registry tabletargets the structural bandno other band would see itcounts stay identicala future survey datetargets the absolute bandstructurally perfectcounts unchangeddelete most of a layertargets the relative bandevery remaining row is validconformance still passesone corruption per band, so a pass proves that banda corruption several bands catch proves nothing about any of them
Targeting one band each is what makes the negative tests diagnostic rather than merely reassuring.

Alternative Approaches or Edge Cases

Gating a container you did not build. For an incoming delivery, widen the advisory band considerably — failing on someone else’s non-conformance helps nobody. What is still worth blocking is registry integrity and geometry decode, because those genuinely break the pipeline downstream.

Per-layer severity. A layer that is critical to the deliverable and one that is supplementary do not deserve the same treatment. Attaching a severity to each rule and each layer costs a little configuration and stops a gate blocking a release over a reference layer nobody uses.

Running the gate on the device. A field device applying a delta can run the same absolute rules before committing, which catches a bad push at the point it arrives. The structural and relative bands are less useful there, because the device does not hold a manifest history.

Troubleshooting

The gate blocks and the artefact is gone

Cause: The script cleaned up the temporary file on failure. Fix: Do not — the un-published artefact is the primary evidence. Leave it in place and name it in the failure message, as the fail helper does.

The relative band fails on the first run

Cause: No previous manifest exists, and absence was treated as a failure. Fix: Treat a missing baseline as a pass with a recorded note, as in the primary script. Only the second run onward can be compared.

A failed gate still published

Cause: set -e was not in force, or a check ran in a subshell whose exit code was discarded. Fix: Check that every band’s exit status reaches the top level; piping a check into grep masks its status unless pipefail is set.

Frequently Asked Questions

Where should the gate live — in the pipeline or beside it?

Beside it, as a script the pipeline calls. That way the same gate can be run by hand against a suspect container, against an incoming delivery, or in a test, without invoking the whole pipeline. Embedding the checks inside the build makes them unreachable for exactly the cases where they are most useful.

How long should a gate take?

Seconds, for the structural and relative bands, which are index reads and metadata queries. The absolute band scales with the rules: an extent check is nearly free because it reads geometry headers, while a validity scan decodes every geometry and is minutes on a large layer. Put the expensive rules behind a flag and run them on release rather than every cycle.

Should the gate ever fix things?

No. A gate that repairs what it finds is a pipeline stage wearing a gate’s clothing, and it removes the signal that something upstream is wrong. Keep repair in its own stage, before the gate, so the gate’s verdict is about the artefact as it will actually be published.