87 lines
3.7 KiB
Python
87 lines
3.7 KiB
Python
"""breeding 迁移套件统一入口（幂等，按依赖顺序执行）。
|
||
|
||
背景：全系统统一修复产生了多支幂等迁移脚本，分散执行易漏、易错序。
|
||
本脚本按「依赖顺序 + 数据库方言矩阵」串起整套迁移，供 deploy migrate 调用。
|
||
|
||
用法（backend/ 目录）：
|
||
python scripts/run_migrations.py # 执行整套迁移
|
||
python scripts/run_migrations.py --check # 预览（仅支持 --check 的脚本预览，其余跳过）
|
||
|
||
方言矩阵：
|
||
fix_breeding_columns.py ORM↔数据库列补齐 全部方言
|
||
schema_align.py 审计列统一新契约 仅 PostgreSQL
|
||
add_unique_constraints.py 部分唯一索引(未软删)并发兜底 仅 PostgreSQL
|
||
migrate_dict_english.py 字典存量中文 label → 英码 value 全部方言
|
||
|
||
说明：
|
||
- PostgreSQL 专属迁移（部分唯一索引、gen_random_uuid 等）仅在
|
||
postgres 方言下运行，其余方言自动跳过并提示；
|
||
- 每支脚本均幂等，可重复执行；--check 仅对声明支持的脚本生效（不支持的跳过，
|
||
避免 --check 误触发 DDL）。
|
||
"""
|
||
import dataclasses
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
BACKEND_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
if BACKEND_DIR not in sys.path:
|
||
sys.path.insert(0, BACKEND_DIR)
|
||
|
||
SCRIPTS_DIR = Path(__file__).resolve().parent
|
||
|
||
|
||
@dataclasses.dataclass
|
||
class Migration:
|
||
name: str
|
||
dialects: set[str] # 适用方言；"*" 表示全部
|
||
desc: str
|
||
check: bool # 是否支持 --check 预览
|
||
|
||
|
||
SUITE: list[Migration] = [
|
||
Migration("fix_breeding_columns.py", {"*"}, "ORM 与数据库列补齐（生成器改字段后同步）", True),
|
||
Migration("schema_align.py", {"postgres"}, "审计列统一新契约（uuid/is_deleted bool/审计 FK）", False),
|
||
Migration("align_columns.py", {"postgres"}, "业务必填列 NOT NULL 对齐（无 NULL 才收紧）", True),
|
||
Migration("add_unique_constraints.py", {"postgres"}, "业务唯一约束落库（部分唯一索引，未软删）", True),
|
||
Migration("migrate_dict_english.py", {"*"}, "字典存量值中文 label 到英码 value（与 bre_dict.sql 对齐）", True),
|
||
Migration("supplement_tree_status_dict.py", {"*"}, "补充单株状态字典 tree_status 初选/重点/保存 三态", True),
|
||
Migration("create_v_trait_value.py", {"postgres"}, "统一性状值视图 v_trait_value（统计消费）", True),
|
||
]
|
||
|
||
|
||
def main() -> None:
|
||
from app.config.setting import settings
|
||
|
||
check = "--check" in sys.argv
|
||
dialect = settings.DATABASE_TYPE
|
||
print(f"[run_migrations] 数据库方言: {dialect} 模式: {'预览 --check' if check else '执行'}")
|
||
|
||
failed: list[str] = []
|
||
for mig in SUITE:
|
||
if "*" not in mig.dialects and dialect not in mig.dialects:
|
||
print(f"[run_migrations] 跳过 {mig.name}（{mig.desc}）: 方言 {dialect} 不支持")
|
||
continue
|
||
if check and not mig.check:
|
||
print(f"[run_migrations] 跳过 {mig.name}（{mig.desc}）: 该脚本不支持 --check 预览")
|
||
continue
|
||
|
||
cmd = [sys.executable, str(SCRIPTS_DIR / mig.name)]
|
||
if check:
|
||
cmd.append("--check")
|
||
print(f"[run_migrations] 执行 {mig.name}（{mig.desc}）...")
|
||
r = subprocess.run(cmd)
|
||
if r.returncode != 0:
|
||
failed.append(mig.name)
|
||
print(f"[run_migrations] {mig.name} 失败，退出码 {r.returncode}")
|
||
|
||
if failed:
|
||
print(f"[run_migrations] 有 {len(failed)} 个迁移失败: {', '.join(failed)}")
|
||
sys.exit(1)
|
||
print("[run_migrations] 全部迁移完成")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|