Xiaoxia AI
95 lines
3.0 KiB
Python
95 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
if str(REPO_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(REPO_ROOT))
|
|
|
|
from packages.adapters.sqlalchemy_impl.models import Base
|
|
|
|
SNAPSHOT_PATH = Path("docs/schema-metadata-snapshot.json")
|
|
|
|
|
|
def column_signature(column) -> dict[str, Any]:
|
|
return {
|
|
"name": column.name,
|
|
"type": str(column.type),
|
|
"nullable": column.nullable,
|
|
"primary_key": column.primary_key,
|
|
"unique": bool(column.unique),
|
|
"index": bool(column.index),
|
|
}
|
|
|
|
|
|
def index_signature(index) -> dict[str, Any]:
|
|
return {
|
|
"name": index.name,
|
|
"columns": [column.name for column in index.columns],
|
|
"unique": index.unique,
|
|
}
|
|
|
|
|
|
def metadata_signature() -> dict[str, Any]:
|
|
tables: dict[str, Any] = {}
|
|
for table_name in sorted(Base.metadata.tables):
|
|
table = Base.metadata.tables[table_name]
|
|
tables[table_name] = {
|
|
"columns": [column_signature(column) for column in table.columns],
|
|
"indexes": sorted(
|
|
[index_signature(index) for index in table.indexes],
|
|
key=lambda item: item["name"] or "",
|
|
),
|
|
"primary_key": [column.name for column in table.primary_key.columns],
|
|
}
|
|
return {"tables": tables}
|
|
|
|
|
|
def write_snapshot(path: Path) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(metadata_signature(), indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
|
|
|
|
|
def check_snapshot(path: Path) -> int:
|
|
expected = json.loads(path.read_text(encoding="utf-8"))
|
|
actual = metadata_signature()
|
|
if actual == expected:
|
|
return 0
|
|
|
|
print(
|
|
"SQLAlchemy metadata drift detected. "
|
|
"If this schema change is intentional, add an Alembic revision and refresh "
|
|
f"{path} with `python scripts/check_schema_metadata.py --write`."
|
|
)
|
|
expected_tables = set(expected.get("tables", {}))
|
|
actual_tables = set(actual.get("tables", {}))
|
|
if expected_tables != actual_tables:
|
|
print(
|
|
f"Table diff: missing={sorted(expected_tables - actual_tables)} added={sorted(actual_tables - expected_tables)}"
|
|
)
|
|
|
|
for table_name in sorted(expected_tables & actual_tables):
|
|
if expected["tables"][table_name] != actual["tables"][table_name]:
|
|
print(f"Changed table: {table_name}")
|
|
return 1
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Check SQLAlchemy metadata drift against committed snapshot.")
|
|
parser.add_argument("--write", action="store_true", help="Refresh the committed metadata snapshot.")
|
|
parser.add_argument("--snapshot", type=Path, default=SNAPSHOT_PATH)
|
|
args = parser.parse_args()
|
|
|
|
if args.write:
|
|
write_snapshot(args.snapshot)
|
|
return 0
|
|
return check_snapshot(args.snapshot)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|