136 lines
5.4 KiB
Python
136 lines
5.4 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
D2 — gen_l0_sql.py
|
|
Erzeugt den kanonischen L0-SQL-Contract (schema-only) aus einem Captured-Schema-Modell.
|
|
|
|
Repräsentiert die aktuelle kanonische Production-Truth. Enthält KEINE:
|
|
data rows, credentials, volatile timestamps, container IDs, host paths, secrets.
|
|
Kein CREATE DATABASE, keine Grant/GRANT-relevanten Owner, keine Umgebungs-Sequenzen.
|
|
|
|
Semantische Overrides (§4) werden als Kommentar-Blöcke dokumentiert:
|
|
last_successful_chunk = timestamp with time zone (Owner-canonical)
|
|
interrupt_reason = text (preserved current contract)
|
|
technical_quality = text nullable default 'VALID'
|
|
"""
|
|
import json
|
|
import sys
|
|
import argparse
|
|
|
|
TYPE_MAP = {
|
|
'bigint': 'bigint', 'integer': 'integer', 'smallint': 'smallint',
|
|
'numeric': 'numeric', 'timestamp with time zone': 'timestamp with time zone',
|
|
'boolean': 'boolean', 'text': 'text', 'date': 'date', 'double precision': 'double precision',
|
|
'jsonb': 'jsonb', 'real': 'real', 'character varying': 'character varying',
|
|
}
|
|
|
|
|
|
def ident(s):
|
|
return '"' + str(s).replace('"', '""') + '"'
|
|
|
|
|
|
def col_def(c):
|
|
typ = TYPE_MAP.get(c.get('type'), c.get('udt', c.get('type', 'text')))
|
|
nullable = '' if not c.get('nullable') else ''
|
|
notnull = ' NOT NULL' if not c.get('nullable') else ''
|
|
dflt = c.get('default')
|
|
default_part = ''
|
|
if dflt:
|
|
default_part = ' DEFAULT ' + dflt
|
|
return f" {ident(c['column'])} {typ}{notnull}{default_part}"
|
|
|
|
|
|
def build_sql(model, dbname="historical"):
|
|
out = []
|
|
out.append("-- ============================================================")
|
|
out.append(f"-- CANONICAL CURRENT PRODUCTION SCHEMA CONTRACT (L0)")
|
|
out.append(f"-- db: {dbname} · schema: public")
|
|
out.append("-- Auto-generated by gen_l0_sql.py from read-only capture")
|
|
out.append("-- Modus: CURRENT VERIFIED PRODUCTION TRUTH — NICHT historische Migration")
|
|
out.append("-- ============================================================")
|
|
out.append("")
|
|
out.append("-- SEMANTIC OVERRIDES (Owner-Canonical)")
|
|
out.append("-- last_successful_chunk = timestamp with time zone (Production + Owner decision)")
|
|
out.append("-- interrupt_reason = text · PRESERVED CURRENT CONTRACT · origin unknown · do not remove")
|
|
out.append("-- technical_quality = text · nullable · default 'VALID' · historical provenance debt OPEN")
|
|
out.append("")
|
|
# Sequenzen (für nextval-Defaults)
|
|
for seq in sorted(model.get("sequences") or []):
|
|
out.append(f"CREATE SEQUENCE IF NOT EXISTS {ident(seq)};")
|
|
if model.get("sequences"):
|
|
out.append("")
|
|
|
|
tables = sorted(model.get("tables") or [])
|
|
cols_by_table = {}
|
|
for c in model.get("columns") or []:
|
|
cols_by_table.setdefault(c["table"], []).append(c)
|
|
|
|
pks = {}
|
|
for pk in model.get("primary_keys") or []:
|
|
cols = sorted(pk.get("columns") or [], key=lambda x: 0)
|
|
pks[pk["table"]] = [pk.get("columns") or []]
|
|
|
|
for t in tables:
|
|
cols = sorted(cols_by_table.get(t, []), key=lambda c: c.get("ordinal", 0))
|
|
out.append(f"-- Table: {t} ({len(cols)} cols)")
|
|
out.append(f"CREATE TABLE IF NOT EXISTS {ident(t)} (")
|
|
body_lines = [col_def(c) for c in cols]
|
|
out.append(",\n".join(body_lines))
|
|
out.append(");")
|
|
out.append("")
|
|
|
|
# PKs
|
|
for t in tables:
|
|
for pk in model.get("primary_keys") or []:
|
|
if pk["table"] == t:
|
|
colstr = ", ".join(ident(c) for c in pk.get("columns") or [])
|
|
out.append(f"ALTER TABLE {ident(t)} ADD CONSTRAINT {ident(pk['constraint'])} PRIMARY KEY ({colstr});")
|
|
out.append("")
|
|
|
|
# UNIQUE
|
|
for uq in sorted(model.get("unique_constraints") or [], key=lambda x: x["table"]):
|
|
colstr = ", ".join(ident(c) for c in uq.get("columns") or [])
|
|
out.append(f"ALTER TABLE {ident(uq['table'])} ADD CONSTRAINT {ident(uq['constraint'])} UNIQUE ({colstr});")
|
|
out.append("")
|
|
|
|
# FK
|
|
for fk in sorted(model.get("foreign_keys") or [], key=lambda x: x["table"]):
|
|
colstr = ", ".join(ident(c) for c in fk.get("columns") or [])
|
|
rcolstr = ", ".join(ident(c) for c in fk.get("ref_columns") or [])
|
|
out.append(f"ALTER TABLE {ident(fk['table'])} ADD CONSTRAINT {ident(fk['constraint'])} FOREIGN KEY ({colstr}) REFERENCES {ident(fk['ref_table'])} ({rcolstr});")
|
|
out.append("")
|
|
|
|
# CHECK - semantische Definition
|
|
for chk in sorted(model.get("check_constraints") or [], key=lambda x: (x["table"], x["constraint"])):
|
|
out.append(f"ALTER TABLE {ident(chk['table'])} ADD CONSTRAINT {ident(chk['constraint'])} CHECK ({chk.get('definition','')});")
|
|
out.append("")
|
|
|
|
# Indexe (non-PK/UNIQUE)
|
|
idx_set = set()
|
|
for pk in model.get("primary_keys") or []:
|
|
idx_set.add(pk["constraint"])
|
|
for uq in model.get("unique_constraints") or []:
|
|
idx_set.add(uq["constraint"])
|
|
for idx in sorted(model.get("indexes") or [], key=lambda x: x["index"]):
|
|
if idx["index"] in idx_set:
|
|
continue
|
|
out.append(f"{idx['definition']};")
|
|
out.append("")
|
|
return "\n".join(out)
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("model")
|
|
ap.add_argument("--out", required=True)
|
|
args = ap.parse_args()
|
|
with open(args.model) as f:
|
|
model = json.load(f)
|
|
sql = build_sql(model)
|
|
with open(args.out, "w") as f:
|
|
f.write(sql)
|
|
print(f"L0 SQL-Contract geschrieben: {args.out} ({len(sql)} B)")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|