79 lines
2.3 KiB
Python
79 lines
2.3 KiB
Python
|
"""数据库迁移模块。
|
|||
|
|
|
|||
|
|
使用方式：
|
|||
|
|
# 在应用启动时自动执行所有迁移
|
|||
|
|
from insurance.db import run_migrations
|
|||
|
|
run_migrations()
|
|||
|
|
|
|||
|
|
# 或手动执行单个迁移
|
|||
|
|
from insurance.db.migrate_001 import migrate
|
|||
|
|
migrate()
|
|||
|
|
"""
|
|||
|
|
import importlib
|
|||
|
|
import pkgutil
|
|||
|
|
import os
|
|||
|
|
|
|||
|
|
|
|||
|
|
def run_migrations():
|
|||
|
|
"""自动执行所有迁移脚本。"""
|
|||
|
|
from flask import current_app
|
|||
|
|
|
|||
|
|
# 检查是否启用自动迁移
|
|||
|
|
if not current_app.config.get("MIGRATION_ENABLED", False):
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
print("[db.migrations] 开始执行数据库迁移...")
|
|||
|
|
|
|||
|
|
# 获取当前目录
|
|||
|
|
current_dir = os.path.dirname(os.path.abspath(__file__))
|
|||
|
|
|
|||
|
|
# 扫描所有迁移脚本（migrate_*.py）
|
|||
|
|
migrations = []
|
|||
|
|
for filename in sorted(os.listdir(current_dir)):
|
|||
|
|
if filename.startswith("migrate_") and filename.endswith(".py"):
|
|||
|
|
migrations.append(filename[:-3]) # 去掉 .py 后缀
|
|||
|
|
|
|||
|
|
# 按顺序执行迁移
|
|||
|
|
for migration_name in migrations:
|
|||
|
|
try:
|
|||
|
|
module = importlib.import_module(f"insurance.db.{migration_name}")
|
|||
|
|
if hasattr(module, "migrate"):
|
|||
|
|
print(f"[db.migrations] 执行迁移: {migration_name}")
|
|||
|
|
module.migrate()
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"[db.migrations] 迁移 {migration_name} 失败: {e}")
|
|||
|
|
# 继续执行下一个迁移，不中断
|
|||
|
|
|
|||
|
|
print("[db.migrations] 迁移执行完成")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def get_migration_status():
|
|||
|
|
"""获取迁移执行状态。"""
|
|||
|
|
from insurance.db.compat import db
|
|||
|
|
from sqlalchemy import text
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
# 检查迁移历史表是否存在
|
|||
|
|
result = db.session.execute(text("""
|
|||
|
|
SELECT EXISTS (
|
|||
|
|
SELECT FROM information_schema.tables
|
|||
|
|
WHERE table_name = 'db_migration_history'
|
|||
|
|
)
|
|||
|
|
"""))
|
|||
|
|
|
|||
|
|
if not result.fetchone()[0]:
|
|||
|
|
return {"status": "no_history_table", "migrations": []}
|
|||
|
|
|
|||
|
|
# 获取已执行的迁移
|
|||
|
|
result = db.session.execute(text("""
|
|||
|
|
SELECT migration_name, executed_at
|
|||
|
|
FROM db_migration_history
|
|||
|
|
ORDER BY executed_at DESC
|
|||
|
|
"""))
|
|||
|
|
|
|||
|
|
migrations = [dict(row) for row in result]
|
|||
|
|
return {"status": "ok", "migrations": migrations}
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
return {"status": "error", "message": str(e)}
|