"""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()