#!/usr/bin/env python3 """Générateur SQL des clés étrangères manquantes — issue #156, livrable B14. Lit deux extractions produites par `scripts/schema/extract_schema.sql` (même format TSV que `scripts/schema/diff_schema.py`) : la **référence** (ce que les migrations produisent sur une base vierge) et la **cible** (la copie de production en cours de bascule). En déduit, sans qu'aucune clé étrangère ne soit écrite en dur ici, ce qu'il reste à poser. Pourquoi rien n'est écrit en dur : le nombre de clés étrangères de la référence a déjà changé deux fois (64 au 08/08, 63 annoncées au runbook §0.7, 69 mesurées le 24/08 sur `development`). Un script qui les énumère serait faux au prochain sprint ; un script qui les dérive reste juste. Même raisonnement pour les divergences de type D2/D3 du `19_DIFF_SCHEMA_PROD.md` : elles ne sont pas nommées ici, elles tombent du critère « colonne portant une clé étrangère à poser dont le type diffère de la référence ». Modes (tous écrivent du SQL sur la sortie standard, à passer au client `mysql`) : types ALTER … MODIFY alignant le type des colonnes porteuses d'une FK manquante sur celui de la référence. À jouer AVANT `migrate`. garde-negatifs SELECT de contrôle : compte les valeurs négatives des colonnes que le mode `types` s'apprête à passer en UNSIGNED. Une seule ligne négative interdit la conversion (perte de donnée silencieuse). orphelins SELECT de contrôle : compte, pour chaque FK manquante, les lignes pointant vers un enregistrement inexistant. Une seule suffit à faire échouer la création en errno 1452. fk-ddl ALTER … ADD CONSTRAINT pour chaque FK manquante. Options : --table T restreint la sortie à la table T (contrôle ciblé avant `migrate`). --sans-entete n'émet pas les commentaires SQL d'en-tête. Idempotence : les quatre modes se calculent à partir de l'écart référence ↔ cible. Une fois l'écart résorbé, ils n'émettent plus aucune instruction — c'est ce qui fait de la seconde passe un no-op, sans garde ad hoc à maintenir. """ import argparse import re import sys from collections import defaultdict # --- lecture du TSV de extract_schema.sql ----------------------------------------- def parse(path): sections = defaultdict(list) cur = None with open(path) as f: for line in f: line = line.rstrip("\n") if line.startswith("###"): cur = line.strip("#") continue if not line.strip() or cur is None: continue sections[cur].append(line.split("\t")) return sections def norm_type(t): """Neutralise les écarts de représentation (même règle que diff_schema.py).""" t = t.strip().lower() t = re.sub(r"\b(tinyint|smallint|mediumint|int|integer|bigint)\(\d+\)", r"\1", t) t = t.replace("integer", "int") return re.sub(r"\s+", " ", t) def build(sections): tables = {r[0]: {"engine": r[1]} for r in sections["TABLES"]} cols = {} for r in sections["COLUMNS"]: tn, cn, ctype, nullable, default, extra = r[0], r[1], r[2], r[3], r[4], r[5] cols[(tn, cn)] = { "type": norm_type(ctype), "raw_type": ctype, "nullable": nullable, "default": default, "extra": extra or "", } fks = {} for r in sections["FKS"]: tn, name, col, rt, rc, dr, ur = r[0], r[1], r[2], r[3], r[4], r[5], r[6] fks.setdefault((tn, name), {"columns": [], "ref_table": rt, "ref_columns": [], "on_delete": dr, "on_update": ur}) fks[(tn, name)]["columns"].append(col) fks[(tn, name)]["ref_columns"].append(rc) return {"tables": tables, "columns": cols, "fks": fks} # --- calcul de l'écart ------------------------------------------------------------- def fks_manquantes(ref, cible, table=None): """FK présentes dans la référence, absentes de la cible. Appariement par (table, colonnes, table référencée) et JAMAIS par nom : les noms de contraintes générés diffèrent d'une base à l'autre, et c'est exactement l'erreur du recalage du 06/08 (#137) qu'il ne faut pas rejouer. """ out = [] for (tn, name), rf in sorted(ref["fks"].items()): if table is not None and tn != table: continue if tn not in cible["tables"]: continue # la table n'existe pas encore : `migrate` la créera avec sa FK if any((tn, c) not in cible["columns"] for c in rf["columns"]): continue # colonne pas encore créée : idem if rf["ref_table"] not in cible["tables"]: continue deja = [p for (ptn, _), p in cible["fks"].items() if ptn == tn and p["columns"] == rf["columns"] and p["ref_table"] == rf["ref_table"]] if not deja: out.append({"table": tn, "name": name, **rf}) return out def colonnes_a_realigner(ref, cible, manquantes): """Colonnes porteuses d'une FK manquante dont le type diverge de la référence. C'est le critère qui couvre D2 (`lessons.course_id` en `int`), D3 (`learning_paths.course_type_id` signé) — et celles qu'on n'a pas encore vues. """ vues, out = set(), [] for fk in manquantes: for i, col in enumerate(fk["columns"]): cle = (fk["table"], col) if cle in vues: continue r, c = ref["columns"].get(cle), cible["columns"].get(cle) if r is None or c is None or r["type"] == c["type"]: continue vues.add(cle) out.append({"table": fk["table"], "column": col, "ref": r, "cible": c, "vers_unsigned": "unsigned" in r["type"] and "unsigned" not in c["type"]}) return out # --- rendu SQL --------------------------------------------------------------------- def q(ident): return "`" + ident.replace("`", "``") + "`" def definition(col): """Rend la définition d'une colonne telle que la référence la décrit.""" bout = [col["raw_type"]] bout.append("NULL" if col["nullable"] == "YES" else "NOT NULL") d = col["default"] if d not in ("__NULL__", None, ""): if re.fullmatch(r"-?\d+(\.\d+)?", d.strip()) or d.strip().lower().startswith("current_timestamp"): bout.append("DEFAULT " + d.strip()) else: bout.append("DEFAULT '" + d.replace("'", "''") + "'") extra = (col["extra"] or "").replace("DEFAULT_GENERATED", "").strip() if extra: bout.append(extra) return " ".join(bout) def main(): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--ref", required=True, help="TSV du schéma de référence") ap.add_argument("--cible", required=True, help="TSV du schéma de la base à basculer") ap.add_argument("--mode", required=True, choices=["types", "garde-negatifs", "orphelins", "fk-ddl"]) ap.add_argument("--table", default=None) ap.add_argument("--sans-entete", action="store_true") a = ap.parse_args() ref, cible = build(parse(a.ref)), build(parse(a.cible)) manquantes = fks_manquantes(ref, cible, a.table) lignes = [] if a.mode == "types": for c in colonnes_a_realigner(ref, cible, manquantes): lignes.append("ALTER TABLE {} MODIFY {} {};".format( q(c["table"]), q(c["column"]), definition(c["ref"]))) elif a.mode == "garde-negatifs": parts = [] for c in colonnes_a_realigner(ref, cible, manquantes): if not c["vers_unsigned"]: continue parts.append( "SELECT '{t}.{c}' AS colonne, COUNT(*) AS valeurs_negatives " "FROM {qt} WHERE {qc} < 0".format( t=c["table"], c=c["column"], qt=q(c["table"]), qc=q(c["column"]))) if parts: lignes.append("\nUNION ALL\n".join(parts) + ";") elif a.mode == "orphelins": parts = [] for fk in manquantes: enfant, parent = q(fk["table"]), q(fk["ref_table"]) jointure = " AND ".join( "p.{} = e.{}".format(q(rc), q(c)) for c, rc in zip(fk["columns"], fk["ref_columns"])) non_nuls = " AND ".join("e.{} IS NOT NULL".format(q(c)) for c in fk["columns"]) parts.append( "SELECT '{t}' AS enfant, '{cols}' AS colonnes, '{rt}' AS parent, " "COUNT(*) AS orphelins FROM {qe} e LEFT JOIN {qp} p ON {j} " "WHERE {nn} AND p.{first} IS NULL".format( t=fk["table"], cols=",".join(fk["columns"]), rt=fk["ref_table"], qe=enfant, qp=parent, j=jointure, nn=non_nuls, first=q(fk["ref_columns"][0]))) if parts: lignes.append("\nUNION ALL\n".join(parts) + ";") elif a.mode == "fk-ddl": for fk in manquantes: cols = ", ".join(q(c) for c in fk["columns"]) rcols = ", ".join(q(c) for c in fk["ref_columns"]) lignes.append( "ALTER TABLE {t} ADD CONSTRAINT {n} FOREIGN KEY ({c}) " "REFERENCES {rt} ({rc}) ON DELETE {od} ON UPDATE {ou};".format( t=q(fk["table"]), n=q(fk["name"]), c=cols, rt=q(fk["ref_table"]), rc=rcols, od=fk["on_delete"], ou=fk["on_update"])) if not a.sans_entete: print("-- cles_etrangeres.py --mode {} : {} instruction(s)".format( a.mode, len(lignes)), file=sys.stderr) for l in lignes: print(l) return 0 if __name__ == "__main__": sys.exit(main())