135 lines
4.1 KiB
Python
135 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
||
"""Vergleicht db/schema.sql mit den SQLAlchemy-Modellen.
|
||
|
||
python3 tools/check-schema.py
|
||
|
||
Die SQL-Datei ist von Hand geschrieben und kann beim nächsten
|
||
Modell-Umbau vergessen werden. Dieser Vergleich findet fehlende oder
|
||
überzählige Tabellen und Spalten - bevor jemand die Datei zum Aufsetzen
|
||
einer neuen Instanz benutzt und sich wundert.
|
||
|
||
Arbeitet rein textlich, ohne SQLAlchemy zu laden: Damit läuft die
|
||
Prüfung auch außerhalb des Containers.
|
||
"""
|
||
|
||
import re
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parent.parent
|
||
MODELS = ROOT / "backend" / "app" / "models.py"
|
||
SCHEMA = ROOT / "db" / "schema.sql"
|
||
|
||
# Tabellen, die nicht aus Modellen stammen
|
||
EXTRA_TABLES = {"alembic_version"}
|
||
|
||
|
||
def from_models() -> dict[str, set[str]]:
|
||
"""__tablename__ und mapped_column-Namen je Klasse einsammeln."""
|
||
source = MODELS.read_text(encoding="utf-8")
|
||
tables: dict[str, set[str]] = {}
|
||
current: str | None = None
|
||
|
||
for line in source.splitlines():
|
||
table = re.match(r'\s*__tablename__\s*=\s*"([^"]+)"', line)
|
||
if table:
|
||
current = table.group(1)
|
||
tables[current] = set()
|
||
continue
|
||
|
||
if current is None:
|
||
continue
|
||
|
||
# Beziehungen sind keine Spalten
|
||
if "relationship(" in line:
|
||
continue
|
||
|
||
column = re.match(r"\s*(\w+):\s*Mapped\[", line)
|
||
if column:
|
||
tables[current].add(column.group(1))
|
||
|
||
return tables
|
||
|
||
|
||
def from_schema() -> dict[str, set[str]]:
|
||
"""CREATE TABLE ... ( ... ) auswerten."""
|
||
source = SCHEMA.read_text(encoding="utf-8")
|
||
tables: dict[str, set[str]] = {}
|
||
|
||
for match in re.finditer(
|
||
r"CREATE TABLE `(\w+)` \((.*?)\n\) ENGINE", source, re.S
|
||
):
|
||
name, body = match.group(1), match.group(2)
|
||
columns = set()
|
||
for line in body.splitlines():
|
||
line = line.strip()
|
||
if line.startswith("--") or not line:
|
||
continue
|
||
# Spaltendefinitionen beginnen mit `name` gefolgt von einem Typ
|
||
col = re.match(r"`(\w+)`\s+[A-Z]", line)
|
||
if col:
|
||
columns.add(col.group(1))
|
||
tables[name] = columns
|
||
|
||
return tables
|
||
|
||
|
||
def main() -> int:
|
||
models = from_models()
|
||
schema = from_schema()
|
||
problems = 0
|
||
|
||
missing = set(models) - set(schema)
|
||
surplus = set(schema) - set(models) - EXTRA_TABLES
|
||
|
||
for name in sorted(missing):
|
||
print(f" FEHLT in schema.sql: Tabelle {name}")
|
||
problems += 1
|
||
for name in sorted(surplus):
|
||
print(f" ÜBERZÄHLIG in schema.sql: Tabelle {name}")
|
||
problems += 1
|
||
|
||
for name in sorted(set(models) & set(schema)):
|
||
only_model = models[name] - schema[name]
|
||
only_schema = schema[name] - models[name]
|
||
for column in sorted(only_model):
|
||
print(f" FEHLT in schema.sql: {name}.{column}")
|
||
problems += 1
|
||
for column in sorted(only_schema):
|
||
print(f" ÜBERZÄHLIG in schema.sql: {name}.{column}")
|
||
problems += 1
|
||
|
||
# Migrationsstand muss zur höchsten Revision passen
|
||
versions = ROOT / "backend" / "alembic" / "versions"
|
||
revisions = sorted(
|
||
m.group(1)
|
||
for f in versions.glob("*.py")
|
||
if (m := re.search(r'^revision:\s*str\s*=\s*"(\w+)"', f.read_text(), re.M))
|
||
)
|
||
head = revisions[-1] if revisions else None
|
||
stamped = re.search(
|
||
r"INSERT INTO `alembic_version`.*VALUES \('(\w+)'\)", SCHEMA.read_text()
|
||
)
|
||
if head and (not stamped or stamped.group(1) != head):
|
||
print(
|
||
f" Migrationsstand passt nicht: schema.sql trägt "
|
||
f"{stamped.group(1) if stamped else 'nichts'}, "
|
||
f"höchste Revision ist {head}"
|
||
)
|
||
problems += 1
|
||
|
||
if problems:
|
||
print(f"\n{problems} Abweichung(en) zwischen Modellen und schema.sql")
|
||
return 1
|
||
|
||
print(
|
||
f"{len(models)} Tabellen, "
|
||
f"{sum(len(c) for c in models.values())} Spalten – "
|
||
f"schema.sql stimmt mit den Modellen überein (Revision {head})"
|
||
)
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|