114 lines
4.1 KiB
Python
114 lines
4.1 KiB
Python
|
"""补齐 breeding 表因生成器重生而缺失的列（create_all 不会 ALTER 已存在表）。
|
||
|
|
|
||
|
|
背景：FastApiAdmin 用 create_all 建表，但模型经生成器重生后新增的字段不会自动落到
|
||
|
|
已存在的表上，导致接口报 `column xxx does not exist`。本脚本自动比对
|
||
|
|
`app.api.v1.module_bre` 下所有模型与数据库实际列，补齐缺失列 + 外键 + 索引。
|
||
|
|
|
||
|
|
用法（在 backend/ 目录下，conda `dpb` 环境）：
|
||
|
|
python scripts/fix_breeding_columns.py # 默认执行补齐
|
||
|
|
python scripts/fix_breeding_columns.py --check # 仅打印将执行的 DDL，不落库
|
||
|
|
|
||
|
|
注意：
|
||
|
|
- 加列统一按 NULL 处理，避免 NOT NULL 在有数据表上执行失败；非空约束请在数据就绪后另行 ALTER。
|
||
|
|
- 每条 DDL 独立提交，单条失败不影响其余。
|
||
|
|
- 需要联网数据库；会读取 app.config.setting 的 DB_URI（本地 dev: dpb/dpb@127.0.0.1:5432/dpb）。
|
||
|
|
"""
|
||
|
|
import argparse
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
|
||
|
|
# 让脚本可直接以 `python scripts/fix_breeding_columns.py` 运行
|
||
|
|
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)
|
||
|
|
|
||
|
|
from sqlalchemy import create_engine, inspect, text
|
||
|
|
from sqlalchemy.engine import Engine
|
||
|
|
|
||
|
|
from app.config.setting import settings
|
||
|
|
from app.core.base_model import MappedBase
|
||
|
|
from app.utils.import_util import ImportUtil
|
||
|
|
|
||
|
|
# 仅处理育种模块（已知 schema 漂移范围）；如需覆盖全库可改为不过滤
|
||
|
|
SCOPE_PREFIX = "app.api.v1.module_bre"
|
||
|
|
|
||
|
|
|
||
|
|
def collect_breeding_models() -> list:
|
||
|
|
all_models = ImportUtil.find_models(MappedBase)
|
||
|
|
return [m for m in all_models if m.__module__.startswith(SCOPE_PREFIX)]
|
||
|
|
|
||
|
|
|
||
|
|
def build_ddl(engine: Engine, models: list) -> list[str]:
|
||
|
|
insp = inspect(engine)
|
||
|
|
ddl: list[str] = []
|
||
|
|
for m in models:
|
||
|
|
table = m.__tablename__
|
||
|
|
try:
|
||
|
|
existing = insp.get_columns(table)
|
||
|
|
except Exception as e:
|
||
|
|
print(f"# skip {table}: {e}")
|
||
|
|
continue
|
||
|
|
existing_names = {c["name"] for c in existing}
|
||
|
|
for col in m.__table__.columns:
|
||
|
|
if col.name in existing_names:
|
||
|
|
continue
|
||
|
|
# 加列统一 NULL，避免 NOT NULL 在有数据表上报错
|
||
|
|
ddl.append(
|
||
|
|
f"ALTER TABLE {table} ADD COLUMN {col.name} "
|
||
|
|
f"{col.type.compile(dialect=engine.dialect)} NULL;"
|
||
|
|
)
|
||
|
|
for fk in col.foreign_keys:
|
||
|
|
fk_name = f"fk_{table}_{col.name}"
|
||
|
|
ondelete = f" ON DELETE {fk.ondelete}" if fk.ondelete else ""
|
||
|
|
tgt_tbl = fk.column.table.name
|
||
|
|
tgt_col = fk.column.name
|
||
|
|
ddl.append(
|
||
|
|
f"ALTER TABLE {table} ADD CONSTRAINT {fk_name} "
|
||
|
|
f"FOREIGN KEY ({col.name}) REFERENCES {tgt_tbl} ({tgt_col}){ondelete};"
|
||
|
|
)
|
||
|
|
if getattr(col, "index", False):
|
||
|
|
idx_name = f"ix_{table}_{col.name}"
|
||
|
|
ddl.append(f"CREATE INDEX {idx_name} ON {table} ({col.name});")
|
||
|
|
return ddl
|
||
|
|
|
||
|
|
|
||
|
|
def main() -> None:
|
||
|
|
parser = argparse.ArgumentParser(description="补齐 breeding 表缺失列")
|
||
|
|
parser.add_argument(
|
||
|
|
"--check", action="store_true", help="仅打印将执行的 DDL，不落库"
|
||
|
|
)
|
||
|
|
args = parser.parse_args()
|
||
|
|
|
||
|
|
models = collect_breeding_models()
|
||
|
|
print(f"# 收集到 {len(models)} 个 breeding 模型，连接: {settings.DB_URI}")
|
||
|
|
|
||
|
|
engine = create_engine(settings.DB_URI)
|
||
|
|
ddl = build_ddl(engine, models)
|
||
|
|
print(f"# 待执行 DDL 共 {len(ddl)} 条")
|
||
|
|
for stmt in ddl:
|
||
|
|
print(" " + stmt)
|
||
|
|
|
||
|
|
if args.check:
|
||
|
|
print("# --check 模式：未执行任何变更。")
|
||
|
|
return
|
||
|
|
|
||
|
|
if not ddl:
|
||
|
|
print("# 无缺失列，无需补齐。")
|
||
|
|
return
|
||
|
|
|
||
|
|
ok = err = 0
|
||
|
|
for stmt in ddl:
|
||
|
|
try:
|
||
|
|
with engine.begin() as conn:
|
||
|
|
conn.execute(text(stmt))
|
||
|
|
print("OK " + stmt)
|
||
|
|
ok += 1
|
||
|
|
except Exception as e:
|
||
|
|
print("ERR " + stmt + " -> " + str(e)[:160])
|
||
|
|
err += 1
|
||
|
|
print(f"# 完成：成功 {ok}，失败 {err}。")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|