85 lines
2.9 KiB
Python
85 lines
2.9 KiB
Python
"""表存在性审计：`baseline_generated.sql` 声明的表 vs MySQL 实际表。
|
|
|
|
与另两个工具互补：
|
|
- `tools/audit_constraints.py`：唯一键 ↔ 文档基线、ORM 映射 ↔ 库列名；
|
|
- `tools/schema_fingerprint.py`：字段、索引与外键的结构指纹；
|
|
- 本脚本：表名集合与列存在性。
|
|
|
|
凭据来自 `.env` 的 `MYSQL_DSN`，不在脚本内硬编码口令。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
from urllib.parse import unquote, urlparse
|
|
|
|
import pymysql
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
SQL_FILE = ROOT / "alembic" / "baseline_generated.sql"
|
|
VERSIONS_DIR = ROOT / "alembic" / "versions"
|
|
|
|
|
|
def expected_tables() -> set[str]:
|
|
text = SQL_FILE.read_text(encoding="utf-8")
|
|
tables = set(re.findall(
|
|
r"CREATE TABLE\s+(?:IF NOT EXISTS\s+)?`?([A-Za-z0-9_]+)`?",
|
|
text,
|
|
flags=re.IGNORECASE,
|
|
))
|
|
for path in VERSIONS_DIR.glob("*.py"):
|
|
tables.update(re.findall(
|
|
r"CREATE TABLE\s+(?:IF NOT EXISTS\s+)?`?([A-Za-z0-9_]+)`?",
|
|
path.read_text(encoding="utf-8"),
|
|
flags=re.IGNORECASE,
|
|
))
|
|
return tables
|
|
|
|
|
|
def mysql_connection() -> pymysql.Connection:
|
|
from app.core.config import get_settings
|
|
|
|
parsed = urlparse(get_settings().mysql_dsn.replace("mysql+asyncmy://", "mysql+pymysql://"))
|
|
return pymysql.connect(
|
|
host=parsed.hostname or "127.0.0.1",
|
|
port=parsed.port or 3306,
|
|
user=unquote(parsed.username or ""),
|
|
password=unquote(parsed.password or ""),
|
|
database=(parsed.path or "/").lstrip("/"),
|
|
charset="utf8mb4",
|
|
)
|
|
|
|
|
|
def main() -> None:
|
|
connection = mysql_connection()
|
|
database = connection.db.decode() if isinstance(connection.db, bytes) else connection.db
|
|
with connection.cursor() as cursor:
|
|
cursor.execute("SELECT table_name FROM information_schema.tables WHERE table_schema=%s", (database,))
|
|
actual = {row[0] for row in cursor.fetchall()} - {"alembic_version"}
|
|
expected = expected_tables()
|
|
missing = expected - actual
|
|
unexpected = actual - expected
|
|
if missing or unexpected:
|
|
raise SystemExit(f"table mismatch missing={sorted(missing)} unexpected={sorted(unexpected)}")
|
|
cursor.execute(
|
|
"SELECT table_name, column_name FROM information_schema.columns WHERE table_schema=%s",
|
|
(database,),
|
|
)
|
|
actual_columns: dict[str, set[str]] = {}
|
|
for table, column in cursor.fetchall():
|
|
actual_columns.setdefault(table, set()).add(column)
|
|
for table in sorted(expected):
|
|
if not actual_columns.get(table):
|
|
raise SystemExit(f"missing columns for {table}")
|
|
connection.close()
|
|
print(f"schema audit passed: {len(expected)} business tables, no missing or unexpected tables")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|