44 lines
1.2 KiB
Python
44 lines
1.2 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
from sqlalchemy import inspect, text
|
||
|
|
|
||
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||
|
|
|
||
|
|
from app.db.session import SessionLocal
|
||
|
|
|
||
|
|
LEGACY_TABLES = [
|
||
|
|
"evaluation_report_exports",
|
||
|
|
"evaluation_records",
|
||
|
|
"session_submissions",
|
||
|
|
"session_orders",
|
||
|
|
"session_runtime_messages",
|
||
|
|
"training_sessions",
|
||
|
|
"case_exam_items",
|
||
|
|
"rubric_templates",
|
||
|
|
"cases",
|
||
|
|
]
|
||
|
|
|
||
|
|
|
||
|
|
def main() -> None:
|
||
|
|
"""旧表清理：在新表链路验证通过后删除不再被业务依赖的旧表。"""
|
||
|
|
with SessionLocal() as db:
|
||
|
|
existing = set(inspect(db.bind).get_table_names()) if db.bind else set()
|
||
|
|
dialect = db.bind.dialect.name if db.bind else ""
|
||
|
|
if dialect == "mysql":
|
||
|
|
db.execute(text("SET FOREIGN_KEY_CHECKS=0"))
|
||
|
|
for table_name in LEGACY_TABLES:
|
||
|
|
if table_name in existing:
|
||
|
|
db.execute(text(f"DROP TABLE `{table_name}`"))
|
||
|
|
print(f"dropped legacy table: {table_name}")
|
||
|
|
if dialect == "mysql":
|
||
|
|
db.execute(text("SET FOREIGN_KEY_CHECKS=1"))
|
||
|
|
db.commit()
|
||
|
|
print("legacy table cleanup completed")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|