Build a tiny schema-diff gate

# /// script
# requires-python = ">=3.10"
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""Build a tiny schema-diff gate.

A compatibility classifier for schema changes (add optional: allowed; add
required without a default: fail; drop required: fail; widen int32 to int64:
allowed; narrow: fail), ten test cases, and the A, B, C trap (a field is a
string in A, removed in B, an integer in C). Each case is checked against the
last version only and against every earlier version.
"""
import argparse, json

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--json", help="Also write the results to this JSON file.")
args = p.parse_args()

WIDER = {("int32", "int64"), ("float32", "float64"), ("int32", "float64")}
def classify(old, new):
    """Can data written with `old` be read with `new` (and the reverse)? Returns a list of problems."""
    problems = []
    for name, f in old.items():
        if name not in new:
            if f.get("required"): problems.append(f"drop required {name}")
        elif new[name]["type"] != f["type"]:
            if (f["type"], new[name]["type"]) not in WIDER: problems.append(f"{name}: {f['type']} -> {new[name]['type']} (narrow or incompatible)")
    for name, f in new.items():
        if name not in old and f.get("required") and "default" not in f: problems.append(f"add required {name} without default")
    return problems
def gate(history, new, transitive):
    targets = history if transitive else history[-1:]
    return [pb for old in targets for pb in classify(old, new)]
F = lambda t, req=False, **kw: {"type": t, "required": req, **kw}
base = {"id": F("int64", True), "qty": F("int32", True)}
cases = [
    ("add optional field", [base], {**base, "note": F("string")}),
    ("add required field with default", [base], {**base, "region": F("string", True, default="EU")}),
    ("add required field without default", [base], {**base, "region": F("string", True)}),
    ("drop optional field", [{**base, "note": F("string")}], base),
    ("drop required field", [base], {"id": F("int64", True)}),
    ("widen int32 to int64", [base], {**base, "qty": F("int64", True)}),
    ("narrow int64 to int32", [base], {**base, "id": F("int32", True)}),
    ("int32 to string", [base], {**base, "qty": F("string", True)}),
    ("no change", [base], dict(base)),
    ("A, B, C trap: note string, removed, then integer", [{**base, "note": F("string")}, base], {**base, "note": F("int64")}),
]
rows = []
for name, history, new in cases:
    last, allv = gate(history, new, False), gate(history, new, True)
    rows.append({"case": name, "last_version_only": "fail" if last else "pass", "all_versions": "fail" if allv else "pass",
                 "reason": "; ".join(allv) or "-"})
for r in rows: print(f"{r['case']:<50} last only: {r['last_version_only']:<5} all versions: {r['all_versions']:<5} {r['reason']}")

if args.json:
    with open(args.json, "w") as f:
        json.dump({"tables": {"cases": rows}}, f)