feat(db): P2 摘除运行时种子 + 脚本化迁移/种子(migrate.py/seed.py)

- Database.initialize() 不再 create_all/seed_data:运行时仅做库连通检查(不建表不灌种子)。
- 新增 scripts/db/migrate.py(封装 alembic upgrade/downgrade)、scripts/db/seed.py(非运行态灌平台种子)。
- 新增 serverdata/seed/(README) 存放种子 JSON。
- 迁移/种子流程:uv run python scripts/db/migrate.py + scripts/db/seed.py,再启动应用。
- 已知待修:seed 内 AGENT_SEED 全局固定 id 唯一冲突(既有缺陷, 待 P2 后续修);pine/角色已能灌入。
This commit is contained in:
Pine
2026-08-24 19:32:41 +08:00
parent 8b0bca36a1
commit 75caab4d39
5 changed files with 85 additions and 14 deletions
+6 -14
View File
@@ -1857,21 +1857,13 @@ class Database:
self.stats = StatsRepository(self.session) self.stats = StatsRepository(self.session)
async def initialize(self) -> None: async def initialize(self) -> None:
"""建表(本实例的 engine)+ 幂等种子(roles 空时才写)。""" """运行时仅保活:不建表、不灌种子(迁移/种子一律走 alembic + scripts/db/seed.py)。
from .db import Base 仅做一次连通性检查,确保库可达。"""
from .seed import seed_data
if self._engine is not None: if self._engine is not None:
async with self._engine.begin() as conn: from sqlalchemy import text
await conn.run_sync(Base.metadata.create_all) async with self._engine.connect() as conn:
# 种子容错:个别幂等保护不严的地方(如 demo 用户固定 agent id)可能触发唯一冲突, await conn.execute(text("SELECT 1"))
# 不能让种子失败拖垮 lifespan 建库,否则 app.state.db 未设置→全部 500 # 建表/种子:由 alembic upgrade + scripts/db/seed.py 在非运行态执行,禁止运行时注入
try:
await seed_data(self.session)
except Exception as e: # noqa: BLE001
import logging
logging.getLogger("servercore").warning("种子数据写入失败(忽略,应用仍可用): %s", e)
await self.session.rollback()
async def close(self) -> None: async def close(self) -> None:
if self._owns_session: if self._owns_session:
+2
View File
@@ -0,0 +1,2 @@
# -*- coding: utf-8 -*-
"""数据库脚本(migrate / seed)—— 非运行态执行。"""
+30
View File
@@ -0,0 +1,30 @@
# -*- coding: utf-8 -*-
"""数据库迁移脚本(非运行态)—— Alembic 封装。
用法:uv run python scripts/db/migrate.py # alembic upgrade head
uv run python scripts/db/migrate.py downgrade # 回退一个版本
在启动应用【之前】执行;禁止在应用启动时迁移/建表。
"""
from __future__ import annotations
import subprocess
import sys
import os
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent.parent
os.chdir(ROOT)
def main() -> None:
cmd = sys.argv[1] if len(sys.argv) > 1 else "upgrade"
if cmd == "downgrade":
args = ["python", "-m", "alembic", "downgrade", "-1"]
else:
args = ["python", "-m", "alembic", "upgrade", "head"]
print(f"执行迁移: {' '.join(args)}")
raise SystemExit(subprocess.call(args, cwd=str(ROOT)))
if __name__ == "__main__":
main()
+43
View File
@@ -0,0 +1,43 @@
# -*- coding: utf-8 -*-
"""数据库种子脚本(非运行态)—— 在迁移后、启动应用前执行。
用法:uv run python scripts/db/seed.py
职责:灌入平台/园区/培训的基础种子数据(角色、权限、用户、默认园区、培训事件课程等)。
全部幂等(按标记行判断,重跑安全)。禁止在应用启动时调用本逻辑。
"""
from __future__ import annotations
import asyncio
import logging
import sys
import os
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent))
from app.infrastructure.repositories import Database
from app.infrastructure.seed import seed_data as platform_seed
from app import config
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
log = logging.getLogger("db.seed")
async def run() -> None:
db = Database()
try:
# 平台种子(角色/权限/用户/政务/OPC/投资人/工作台等,幂等)
await platform_seed(db.session)
await db.session.commit()
log.info("平台种子完成:%s", config.DATABASE_URL)
# TODO(P2/P3 后):园区默认园区+企业、培训事件课程 经 serverdata/seed/*.json 灌入(当前由各自子应用迁移后接入)
except Exception as e: # noqa: BLE001
await db.session.rollback()
log.error("种子失败:%s", e)
raise
finally:
await db.close()
if __name__ == "__main__":
asyncio.run(run())
+4
View File
@@ -0,0 +1,4 @@
# serverdata/seed — 种子 JSON 目录
数据库种子数据(角色/权限/用户/园区/培训等)以 JSON 存放于此,经 `scripts/db/seed.py` 在迁移后灌入。
禁止在应用启动时注入种子。