Transform+load: consolidate all insurance lines into policies (step 3)
migration/transform_policies.py folds every insurance Access table into one `policies` table (policy_types discriminator) plus child tables, via a per-table declarative mapping that absorbs the column-name variance (num_id/numer_id, no_poliza/poliza, p_neta/prima_neta/prima1). Any source column not explicitly modeled — the type-specific coverage amounts — is preserved verbatim in coveragesJson, so consolidation loses nothing. Unpivots the hardcoded repeated slots: 4 payment installments (c_1er_pago + pago_subsec x3), up to 3 vehicles (auto tables + MCA2), up to 3 named insured drivers (MCA2 + LICENCIAS). Also loads BENEF -> policy_beneficiaries (by policy number), DATOS -> claims, AJUSTADORES(+ATLAS) -> adjusters, and builds policy_types + insurance_providers lookups. Loaded/validated (dev): 2378 policies (AUTO 1307 / MULT 760 / LICENCIAS 306 / M_EMPR 5; 10 skipped for unresolved customer FK, 0 orphans), 4678 installments, 1110 vehicles, 513 drivers, 126 beneficiaries, 1 claim, 15 providers, 17 adjusters — all child FKs verified 0 orphans. Spot-checked a customer carrying both a utility property and MULT policies (the unified cross-line view). run_all.py: add policies to the ordered pipeline. Customer FK resolves through insurance customer_legacy_refs, so this runs after customers. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.8
@@ -33,6 +33,7 @@ PY = sys.executable # the venv python running this orchestrator
|
||||
STEPS = [
|
||||
"transform_customers.py", # customers + customer_legacy_refs (everything FKs to these)
|
||||
"transform_properties.py", # properties + services + trust accounts
|
||||
"transform_policies.py", # policies + installments/vehicles/drivers/benef/claims/adjusters
|
||||
]
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,391 @@
|
||||
"""
|
||||
Migration plan step 3 (policies): consolidate every insurance line into one
|
||||
`policies` table + child tables, resolving each policy's customer FK through
|
||||
the insurance customer_legacy_refs (run AFTER transform_customers.py).
|
||||
|
||||
One Access table per line of business becomes rows in `policies` tagged by a
|
||||
policy_types discriminator, plus:
|
||||
- policy_payment_installments : the 4 hardcoded payment slots, unpivoted
|
||||
- vehicles : the up-to-3 hardcoded vehicle slots
|
||||
- insured_drivers : the up-to-3 hardcoded named-insured slots
|
||||
- policy_beneficiaries : from BENEF (by policy number)
|
||||
- claims : from DATOS (siniestros)
|
||||
- adjusters : from AJUSTADORES / AJUSTADORESATLAS
|
||||
|
||||
Design (validated against staged data):
|
||||
- Customer link column varies (num_id / numer_id); ~10 policy customer ids
|
||||
across all tables don't resolve to a loaded insurance customer -> those
|
||||
policies are skipped and counted (required FK).
|
||||
- Payment slots: c_1er_pago is the first amount, pago_subsec the recurring
|
||||
amount for slots 2-4; efectivo is a cash flag, no_cheque the check ref.
|
||||
- Any source column not explicitly modeled (coverage amounts: edificio,
|
||||
contenidos, robo, cristales, ...) is preserved verbatim in coveragesJson,
|
||||
so nothing is lost in consolidation.
|
||||
- docs_1/docs_2/foto1 (LONGBINARY) are skipped — documents are step 4.
|
||||
|
||||
Run: ./.venv/bin/python transform_policies.py --env dev
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from decimal import Decimal, InvalidOperation
|
||||
from pathlib import Path
|
||||
|
||||
import pandas as pd
|
||||
|
||||
from dbenv import connect, env_arg
|
||||
|
||||
STG = Path(__file__).parent / "output" / "stg_seguros"
|
||||
LEGACY_DB = "SEGUROS 16_be"
|
||||
NULL = "∅"
|
||||
NOW = datetime.now(timezone.utc).replace(tzinfo=None)
|
||||
_ADDR = {"calle_y_no", "col", "edo", "tel", "cp", "_row_num", "nombre_aseg", "observaciones"}
|
||||
|
||||
|
||||
def s(v):
|
||||
if v is None or pd.isna(v):
|
||||
return None
|
||||
v = str(v).strip()
|
||||
return None if v in ("", NULL, "0000-00-00") else v
|
||||
|
||||
|
||||
def norm_id(v):
|
||||
v = s(v)
|
||||
if v is None:
|
||||
return None
|
||||
if v.endswith(".0"):
|
||||
v = v[:-2]
|
||||
return v
|
||||
|
||||
|
||||
def dec(v):
|
||||
v = s(v)
|
||||
if v is None:
|
||||
return None
|
||||
try:
|
||||
return Decimal(v.replace(",", ""))
|
||||
except (InvalidOperation, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
def dt(v):
|
||||
v = s(v)
|
||||
if v is None:
|
||||
return None
|
||||
d = pd.to_datetime(v, errors="coerce")
|
||||
return None if pd.isna(d) else d.to_pydatetime()
|
||||
|
||||
|
||||
def cur(v):
|
||||
v = (s(v) or "").upper()
|
||||
if v.startswith("DOL") or v.startswith("USD") or "DOLLAR" in v:
|
||||
return "USD"
|
||||
if v.startswith("PES") or v.startswith("MXN") or v.startswith("MN"):
|
||||
return "MXN"
|
||||
return "MXN"
|
||||
|
||||
|
||||
def truthy(v):
|
||||
return (s(v) or "0").lower() in {"1", "-1", "true", "si", "sí", "yes", "x"}
|
||||
|
||||
|
||||
# --- per-table config ------------------------------------------------------- #
|
||||
# fields: policy column -> source column. installments: list of slot dicts.
|
||||
# vehicles: 'trip_underscore' | 'single' | 'mca2' | None. drivers: 'mca2' |
|
||||
# 'licencias' | None.
|
||||
HOME_INST = [
|
||||
dict(seq=1, amt="c_1er_pago", cu="moned", d="fecha_pago", ck="no_cheque", cash="efectivo"),
|
||||
dict(seq=2, amt="pago_subsec", cu="moned_2", d="fecha_pago_2", ck="no_cheque_2", cash="efectivo_2"),
|
||||
dict(seq=3, amt="pago_subsec", cu="moned_3", d="fecha_pago_3", ck="no_cheque_3", cash="efectivo_3"),
|
||||
dict(seq=4, amt="pago_subsec", cu="moned_4", d="fecha_pago_4", ck="no_cheque_4", cash="efectivo_4"),
|
||||
]
|
||||
HOME_FIELDS = dict(polno="no_poliza", agent="agent", comp="comp", desde="desde", hasta="hasta",
|
||||
forma="forma_pago", curcol="moned", pneta="p_neta", dpol="d_pol", com="com",
|
||||
liquidada="liquidada", numliq="num_liquidacion", fliq="f_liquida1", renov="renovacion")
|
||||
AUTO_SINGLE_INST = [dict(seq=1, amt="total", cu="moneda", d="fecha_pago", ck="no_cheque", cash="efectivo")]
|
||||
|
||||
CONFIGS = {
|
||||
"incendio": dict(ptype="INCENDIO", idcol="num_id", fields={**HOME_FIELDS, "curcol": "moneda"},
|
||||
inst=HOME_INST, veh=None, drv=None),
|
||||
"mult": dict(ptype="MULT", idcol="num_id", fields=HOME_FIELDS, inst=HOME_INST, veh=None, drv=None),
|
||||
"m_empr": dict(ptype="M_EMPR", idcol="num_id", fields=HOME_FIELDS, inst=HOME_INST, veh=None, drv=None),
|
||||
"tabla_autos": dict(ptype="AUTO", idcol="numer_id",
|
||||
fields=dict(polno="no_poliza", agent="agent", comp="comp", desde="desde",
|
||||
hasta="hasta", forma="forma_pago", curcol="moneda", pneta="prima_neta",
|
||||
dpol="d_poliza", total="total", liquidada="liquidada",
|
||||
numliq="num_liquidacion", fliq="f_liquida1", renov="renovacion"),
|
||||
inst=AUTO_SINGLE_INST, veh="trip_underscore", drv=None),
|
||||
"tabla_autos_ampl": dict(ptype="AUTO", idcol="numer_id",
|
||||
fields=dict(polno="no_poliza", agent="agent", comp="comp", desde="desde",
|
||||
hasta="hasta", forma="forma_pago", curcol="moneda", pneta="prima_neta",
|
||||
dpol="d_poliza", total="total", liquidada="liquidada",
|
||||
numliq="num_liquidacion", fliq="f_liquida1", renov="renovacion"),
|
||||
inst=AUTO_SINGLE_INST, veh="single", drv=None),
|
||||
"tabla_autos_limit": dict(ptype="AUTO", idcol="numer_id",
|
||||
fields=dict(polno="no_poliza", agent="agent", comp="comp", desde="desde",
|
||||
hasta="hasta", forma="forma_pago", curcol="moneda", pneta="prima_neta",
|
||||
dpol="d_poliza", total="total", liquidada="liquidada", renov="renovacion"),
|
||||
inst=AUTO_SINGLE_INST, veh="single", drv=None),
|
||||
"tabla_autos_ampl_r": dict(ptype="AUTO", idcol="numer_id",
|
||||
fields=dict(polno="no_poliza", agent="agent", comp="comp", desde="desde",
|
||||
hasta="hasta", forma="forma_pago", curcol="moneda", pneta="prima_neta",
|
||||
dpol="d_poliza", com="com", liquidada="liquidada",
|
||||
numliq="num_liquidacion", fliq="f_liquida1", renov="renovacion"),
|
||||
inst=[dict(seq=1, amt="c_1er_pago", cu="moneda", d="fecha_pago", ck="no_cheque", cash="efectivo"),
|
||||
dict(seq=2, amt="pago_subsec", cu="moned_2", d="fecha_pago_2", ck="no_cheque_2", cash="efectivo_2")],
|
||||
veh="single", drv=None),
|
||||
"tabla_autos_rc_r": dict(ptype="AUTO", idcol="numer_id",
|
||||
fields=dict(polno="no_poliza", agent="agent", comp="comp", desde="desde",
|
||||
hasta="hasta", forma="forma_pago", curcol="moneda", pneta="prima_neta",
|
||||
dpol="d_poliza", com="com", liquidada="liquidada",
|
||||
numliq="num_liquidacion", fliq="f_liquida1", renov="renovacion"),
|
||||
inst=[dict(seq=1, amt="c_1er_pago", cu="moneda", d="fecha_pago", ck="no_cheque", cash="efectivo"),
|
||||
dict(seq=2, amt="pago_subsec", cu="moned_2", d="fecha_pago_2", ck="no_cheque_2", cash="efectivo_2")],
|
||||
veh="single", drv=None),
|
||||
"mca2": dict(ptype="AUTO", idcol="numer_id",
|
||||
fields=dict(polno="poliza", agent="agent", comp="comp", desde="desde", hasta="hasta",
|
||||
forma="forma_pago", curcol="moned", pneta="prima1", dpol="d_poliza1",
|
||||
liquidada="liquidada", numliq="nuliquida", fliq="f_liquida1", renov="renovacion"),
|
||||
inst=[dict(seq=1, amt="c_1er_pago", cu="moned", d="fecha_pago", ck="no_cheque", cash="efectivo")],
|
||||
veh="mca2", drv="mca2"),
|
||||
"licencias": dict(ptype="LICENCIAS", idcol="numer_id",
|
||||
fields=dict(polno="no_poliza", agent="agent", comp="comp", desde="desde", hasta="hasta",
|
||||
forma="forma_pago", curcol="moneda", pneta="prima_neta", dpol="d_poliza",
|
||||
total="total", liquidada="liquidada", numliq="num_liquidacion",
|
||||
fliq="f_liquida1", renov="renovacion"),
|
||||
inst=[dict(seq=1, amt="total", cu="moneda", d="fecha_pago", ck="no_cheque", cash="efectivo")],
|
||||
veh=None, drv="licencias"),
|
||||
}
|
||||
|
||||
|
||||
def load(name):
|
||||
df = pd.read_parquet(STG / f"{name}.parquet").sort_values("_row_num").reset_index(drop=True)
|
||||
df = df[[c for c in df.columns if c != "_legacy_source_table"]].copy()
|
||||
for c in df.columns:
|
||||
if c != "_row_num":
|
||||
df[c] = df[c].astype("string").str.strip()
|
||||
return df
|
||||
|
||||
|
||||
def main():
|
||||
env = env_arg()
|
||||
conn = connect(env)
|
||||
print(f"[policies] target env: {env}")
|
||||
c = conn.cursor()
|
||||
|
||||
c.execute("SELECT legacyId, customerId FROM customer_legacy_refs WHERE sourceSystem='insurance'")
|
||||
cust = {r[0]: r[1] for r in c.fetchall()}
|
||||
|
||||
policies, insts, vehicles, drivers = [], [], [], []
|
||||
polno_to_id = {} # policy number -> a policyId (for BENEF/DATOS linking)
|
||||
providers, ptypes = set(), set()
|
||||
skipped = 0
|
||||
|
||||
for table, cfg in CONFIGS.items():
|
||||
df = load(table)
|
||||
F = cfg["fields"]
|
||||
ptypes.add(cfg["ptype"])
|
||||
consumed = {cfg["idcol"], *F.values()}
|
||||
for slot in cfg["inst"]:
|
||||
consumed |= {slot["amt"], slot["cu"], slot["d"], slot["ck"], slot["cash"]}
|
||||
|
||||
for _, row in df.iterrows():
|
||||
cid = cust.get(norm_id(row[cfg["idcol"]]))
|
||||
if not cid:
|
||||
skipped += 1
|
||||
continue
|
||||
pid = str(uuid.uuid4())
|
||||
comp = s(row.get(F.get("comp", ""), None)) if F.get("comp") else None
|
||||
if comp:
|
||||
providers.add(comp)
|
||||
polno = s(row.get(F["polno"])) if F.get("polno") else None
|
||||
if polno:
|
||||
polno_to_id[polno] = pid
|
||||
|
||||
# coverages = everything not explicitly consumed / address / doc-ish
|
||||
cov = {}
|
||||
for col in df.columns:
|
||||
if col in consumed or col in _ADDR:
|
||||
continue
|
||||
if col.startswith(("marca", "modelo", "carroceria", "motor", "placa", "clase",
|
||||
"auto_id", "a_o", "edo", "name", "birth", "edad", "lic",
|
||||
"ocup", "sex", "nombre_aseg_", "docs_", "foto")):
|
||||
continue
|
||||
val = s(row[col])
|
||||
if val is not None:
|
||||
cov[col] = val
|
||||
|
||||
policies.append((
|
||||
pid, polno or "(SIN POLIZA)", cid, cfg["ptype"], comp,
|
||||
s(row.get(F.get("agent", ""))) if F.get("agent") else None,
|
||||
dt(row.get(F.get("desde", ""))) if F.get("desde") else None,
|
||||
dt(row.get(F.get("desde", ""))) if F.get("desde") else None,
|
||||
dt(row.get(F.get("hasta", ""))) if F.get("hasta") else None,
|
||||
dec(row.get(F.get("pneta", ""))) if F.get("pneta") else None,
|
||||
dec(row.get(F.get("dpol", ""))) if F.get("dpol") else None,
|
||||
dec(row.get(F.get("com", ""))) if F.get("com") else None,
|
||||
dec(row.get(F.get("total", ""))) if F.get("total") else None,
|
||||
cur(row.get(F.get("curcol", ""))) if F.get("curcol") else "MXN",
|
||||
s(row.get("observaciones")),
|
||||
json.dumps(cov, ensure_ascii=False) if cov else None,
|
||||
1 if truthy(row.get(F.get("liquidada", ""))) else 0,
|
||||
s(row.get(F.get("numliq", ""))) if F.get("numliq") else None,
|
||||
dt(row.get(F.get("fliq", ""))) if F.get("fliq") else None,
|
||||
LEGACY_DB, table, str(int(row["_row_num"])), NOW,
|
||||
))
|
||||
|
||||
# installments
|
||||
for slot in cfg["inst"]:
|
||||
amt = dec(row.get(slot["amt"]))
|
||||
pdate = dt(row.get(slot["d"]))
|
||||
if amt is None and pdate is None:
|
||||
continue
|
||||
insts.append((str(uuid.uuid4()), pid, slot["seq"], amt,
|
||||
cur(row.get(slot["cu"])), pdate, s(row.get(slot["ck"])),
|
||||
1 if truthy(row.get(slot["cash"])) else 0))
|
||||
|
||||
# vehicles
|
||||
def add_vehicle(make, model, body, engine, plate, year=None, state=None):
|
||||
if not any([make, model, plate, engine]):
|
||||
return
|
||||
vehicles.append((str(uuid.uuid4()), cid, pid, make, model, year, body,
|
||||
engine, plate, state, table, str(int(row["_row_num"]))))
|
||||
if cfg["veh"] == "trip_underscore":
|
||||
for i in ("1", "2", "3"):
|
||||
add_vehicle(s(row.get(f"marca_{i}")), s(row.get(f"modelo_{i}")),
|
||||
s(row.get(f"carroceria_{i}")), s(row.get(f"motor_{i}")),
|
||||
s(row.get(f"placa_{i}")))
|
||||
elif cfg["veh"] == "single":
|
||||
add_vehicle(s(row.get("marca")), s(row.get("modelo")), s(row.get("carroceria")),
|
||||
s(row.get("motor")), s(row.get("placa")))
|
||||
elif cfg["veh"] == "mca2":
|
||||
for i in ("1", "2", "3"):
|
||||
add_vehicle(s(row.get(f"marca{i}")), s(row.get(f"modelo{i}")),
|
||||
s(row.get(f"clase{i}")), s(row.get(f"auto_id{i}")),
|
||||
s(row.get(f"placa{i}")), year=s(row.get(f"a_o{i}")),
|
||||
state=s(row.get(f"edo{i}")))
|
||||
|
||||
# drivers
|
||||
def add_driver(name, birth=None, sex=None, occ=None, lic=None, lstate=None):
|
||||
if not name and not lic:
|
||||
return
|
||||
drivers.append((str(uuid.uuid4()), pid, name, dt(birth), sex, occ, lic, lstate))
|
||||
if cfg["drv"] == "mca2":
|
||||
for i in ("1", "2", "3"):
|
||||
add_driver(s(row.get(f"name{i}")), s(row.get(f"birth{i}")),
|
||||
s(row.get(f"sex{i}")), s(row.get(f"ocup{i}")),
|
||||
s(row.get(f"lic{i}")), s(row.get(f"licedo{i}")))
|
||||
elif cfg["drv"] == "licencias":
|
||||
for i in ("1", "2", "3"):
|
||||
add_driver(s(row.get(f"nombre_aseg_{i}")), lic=s(row.get(f"lic_{i}")),
|
||||
lstate=s(row.get(f"edo_{i}")))
|
||||
|
||||
# beneficiaries (BENEF -> by policy number)
|
||||
benef = load("benef")
|
||||
benes = []
|
||||
for _, r in benef.iterrows():
|
||||
pid = polno_to_id.get(s(r["bpoliza"]))
|
||||
if pid:
|
||||
benes.append((str(uuid.uuid4()), pid, s(r["bnombre"]), s(r["bdireccion"]),
|
||||
s(r["btel"]), s(r["bemail"])))
|
||||
|
||||
# adjusters (AJUSTADORES + ATLAS)
|
||||
adj_rows, adj_by_name = [], {}
|
||||
for t in ("ajustadores", "ajustadoresatlas"):
|
||||
for _, r in load(t).iterrows():
|
||||
aid = str(uuid.uuid4())
|
||||
name = s(r["nombre"])
|
||||
adj_rows.append((aid, s(r["comp"]), s(r["ciudad"]), name, s(r["tel"]), s(r["beeper"])))
|
||||
if name:
|
||||
adj_by_name.setdefault(name.upper(), aid)
|
||||
|
||||
# claims (DATOS)
|
||||
claims = []
|
||||
for _, r in load("datos").iterrows():
|
||||
pid = polno_to_id.get(s(r["poliza"]))
|
||||
if not pid:
|
||||
continue
|
||||
adjid = adj_by_name.get((s(r["ajustador"]) or "").upper())
|
||||
claims.append((str(uuid.uuid4()), pid, s(r["tipo"]), dt(r["fecha_siniestro"]),
|
||||
dt(r["fecha_reportado"]), s(r["descripcion"]), adjid,
|
||||
dec(r["cantidad_reclamada"]), dec(r["cantidad_pactada"]),
|
||||
dt(r["fecha_cheque"]), s(r["num_cheque"]),
|
||||
1 if truthy(r["concluido"]) else 0, s(r["resolucion"])))
|
||||
|
||||
# --- write (children first on truncate) ---
|
||||
c.execute("SET FOREIGN_KEY_CHECKS=0")
|
||||
for t in ("policy_payment_installments", "insured_drivers", "policy_beneficiaries",
|
||||
"claims", "vehicles", "policy_documents", "policies", "policy_types",
|
||||
"insurance_providers", "adjusters"):
|
||||
c.execute(f"TRUNCATE TABLE {t}")
|
||||
c.execute("SET FOREIGN_KEY_CHECKS=1")
|
||||
|
||||
ptype_ids = {n: str(uuid.uuid4()) for n in ptypes}
|
||||
c.executemany("INSERT INTO policy_types (id,name) VALUES (%s,%s)",
|
||||
[(i, n) for n, i in ptype_ids.items()])
|
||||
prov_ids = {n: str(uuid.uuid4()) for n in providers}
|
||||
c.executemany("INSERT INTO insurance_providers (id,name) VALUES (%s,%s)",
|
||||
[(i, n) for n, i in prov_ids.items()])
|
||||
c.executemany("INSERT INTO adjusters (id,company,city,name,phone,beeper) VALUES (%s,%s,%s,%s,%s,%s)", adj_rows)
|
||||
|
||||
# patch policyType/provider FKs into policy tuples
|
||||
pol_cols = ("id,policyNumber,customerId,policyTypeId,insuranceProviderId,agentName,policyDate,"
|
||||
"policyFrom,policyTo,netPremium,policyFee,commission,total,currency,observations,"
|
||||
"coveragesJson,liquidated,liquidationNumber,liquidationDate,legacySourceDb,"
|
||||
"legacySourceTable,legacyId,updatedAt")
|
||||
fixed = []
|
||||
for p in policies:
|
||||
p = list(p)
|
||||
p[3] = ptype_ids.get(p[3]) # ptype name -> policyTypeId
|
||||
p[4] = prov_ids.get(p[4]) # provider name -> insuranceProviderId
|
||||
fixed.append(tuple(p))
|
||||
ph = ",".join(["%s"] * 23)
|
||||
c.executemany(f"INSERT INTO policies ({pol_cols}) VALUES ({ph})", fixed)
|
||||
|
||||
c.executemany("INSERT INTO policy_payment_installments "
|
||||
"(id,policyId,sequence,amount,currency,paidDate,checkNumber,isCash) "
|
||||
"VALUES (%s,%s,%s,%s,%s,%s,%s,%s)", insts)
|
||||
c.executemany("INSERT INTO vehicles (id,customerId,policyId,make,model,modelYear,bodyType,"
|
||||
"engineNumber,licensePlate,stateCode,legacySourceTable,legacyId) "
|
||||
"VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", vehicles)
|
||||
c.executemany("INSERT INTO insured_drivers (id,policyId,fullName,birthDate,sex,occupation,"
|
||||
"licenseNumber,licenseState) VALUES (%s,%s,%s,%s,%s,%s,%s,%s)", drivers)
|
||||
c.executemany("INSERT INTO policy_beneficiaries (id,policyId,name,address,phone,email) "
|
||||
"VALUES (%s,%s,%s,%s,%s,%s)", benes)
|
||||
c.executemany("INSERT INTO claims (id,policyId,claimType,incidentDate,reportedDate,description,"
|
||||
"adjusterId,claimedAmount,settledAmount,settlementDate,checkNumber,resolved,"
|
||||
"resolutionNotes) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", claims)
|
||||
conn.commit()
|
||||
|
||||
# --- report ---
|
||||
def count(t):
|
||||
c.execute(f"SELECT COUNT(*) FROM {t}"); return c.fetchone()[0]
|
||||
c.execute("SELECT pt.name, COUNT(*) FROM policies p JOIN policy_types pt ON p.policyTypeId=pt.id "
|
||||
"GROUP BY pt.name ORDER BY 2 DESC")
|
||||
by_type = c.fetchall()
|
||||
c.execute("SELECT COUNT(*) FROM policies p LEFT JOIN customers c ON p.customerId=c.id WHERE c.id IS NULL")
|
||||
orphans = c.fetchone()[0]
|
||||
print("=== Policies load complete ===")
|
||||
print(f" skipped (unresolved customer): {skipped}")
|
||||
print(f" -> policies : {count('policies')}")
|
||||
for n, k in by_type:
|
||||
print(f" {n:12} {k}")
|
||||
print(f" -> payment_installments : {count('policy_payment_installments')}")
|
||||
print(f" -> vehicles : {count('vehicles')}")
|
||||
print(f" -> insured_drivers : {count('insured_drivers')}")
|
||||
print(f" -> policy_beneficiaries : {count('policy_beneficiaries')}")
|
||||
print(f" -> claims : {count('claims')}")
|
||||
print(f" -> policy_types : {count('policy_types')}")
|
||||
print(f" -> insurance_providers : {count('insurance_providers')}")
|
||||
print(f" -> adjusters : {count('adjusters')}")
|
||||
print(f" orphan policies (bad customer FK): {orphans}")
|
||||
assert orphans == 0, "policy customer FK invariant failed"
|
||||
print(" validation: OK")
|
||||
conn.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user