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