ci: detect sqlalchemy schema drift

.gitea/workflows/ci-cd.yml

@@ -61,9 +61,9 @@ jobs:
 
       - name: Run code quality checks
         run: |
-          python -m compileall -q alembic apps packages tests
-          python -m black --check alembic apps packages tests
-          python -m isort --check-only alembic apps packages tests
+          python -m compileall -q alembic apps packages tests scripts/check_schema_metadata.py
+          python -m black --check alembic apps packages tests scripts/check_schema_metadata.py
+          python -m isort --check-only alembic apps packages tests scripts/check_schema_metadata.py
           python -m flake8 apps packages tests --count --statistics
           bandit -r apps packages -q
 
@@ -73,6 +73,7 @@ jobs:
             python -m alembic upgrade head --sql > /tmp/alembic-upgrade.sql
           test -s /tmp/alembic-upgrade.sql
           grep -q "Running upgrade" /tmp/alembic-upgrade.sql
+          python scripts/check_schema_metadata.py
 
       - name: Run tests
         run: |

.github/workflows/ci-cd.yml

@@ -61,9 +61,9 @@ jobs:
 
       - name: Run code quality checks
         run: |
-          python -m compileall -q alembic apps packages tests
-          python -m black --check alembic apps packages tests
-          python -m isort --check-only alembic apps packages tests
+          python -m compileall -q alembic apps packages tests scripts/check_schema_metadata.py
+          python -m black --check alembic apps packages tests scripts/check_schema_metadata.py
+          python -m isort --check-only alembic apps packages tests scripts/check_schema_metadata.py
           python -m flake8 apps packages tests --count --statistics
           bandit -r apps packages -q
 
@@ -73,6 +73,7 @@ jobs:
             python -m alembic upgrade head --sql > /tmp/alembic-upgrade.sql
           test -s /tmp/alembic-upgrade.sql
           grep -q "Running upgrade" /tmp/alembic-upgrade.sql
+          python scripts/check_schema_metadata.py
 
       - name: Run tests
         run: |

docs/SCHEMA-MAINLINE.md

@@ -26,6 +26,7 @@ staging / production 不允许依赖 `Base.metadata.create_all()` 建表；表
 - staging health endpoint 已确认 healthy
 - `develop` 部署 job 已确认 succeeded
 - CI 已增加 Alembic offline upgrade SQL 生成检查，验证 migration 链可加载并能生成 `upgrade head` SQL
+- CI 已增加 SQLAlchemy metadata snapshot drift 检查，防止改 models 后忘记同步 Alembic revision / schema snapshot
 
 ## 已废弃入口
 
@@ -52,12 +53,22 @@ staging / production 不允许依赖 `Base.metadata.create_all()` 建表；表
 7. staging / production 只允许通过 Alembic 升级 schema。
 8. `AUTO_CREATE_SCHEMA` 只能作为开发/测试兜底，staging / production 会被代码级 guard 阻止开启。
 
+## Schema Drift 检查
+
+CI 使用 `scripts/check_schema_metadata.py` 对比当前 SQLAlchemy metadata 和 `docs/schema-metadata-snapshot.json`。
+
+如果 schema 变更是有意的，必须同时：
+
+1. 修改 `packages/adapters/sqlalchemy_impl/models.py`。
+2. 新增 Alembic revision。
+3. 执行 `python scripts/check_schema_metadata.py --write` 刷新 snapshot。
+4. 提交 models、revision、snapshot 三者。
+
 ## 下一步
 
 后续 schema 工作应继续推进：
 
-1. 新增 revision drift 检查，防止 models 已改但未生成 migration。
-2. 对 production 首次接入 Alembic 前，先执行只读 schema/数据备份检查。
+1. 对 production 首次接入 Alembic 前，先执行只读 schema/数据备份检查。
 
 ## 禁止事项
 

scripts/check_schema_metadata.py

@@ -0,0 +1,94 @@
+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())