feat(db): 引入 Alembic 取代裸 DDL 迁移
- alembic init + env.py 接 settings/models metadata（asyncpg->psycopg2 同步 URL） - 补 3 个 CheckConstraint 到 models（原只在裸 DDL） - 离线生成初始迁移（31 表+约束+95 索引，全 sa.* 通用类型） - init_db 用 _run_alembic_migrations（自动基线+upgrade head）替换 create_tables+ensure_schema_updates（删 92 行裸 DDL） - 删破坏性 migrate_db.py（drop_all） - 既有 DB 首次启动自动 stamp 基线，无需手动 Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -7,9 +7,44 @@ from shared.models.database import User, Role, Permission, UserRole, RolePermiss
|
||||
from shared.services.auth_service import get_password_hash
|
||||
from shared.config.settings import settings
|
||||
from shared.utils.logger import get_logger
|
||||
from alembic.config import Config
|
||||
from alembic import command
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
_ALEMBIC_INI = Path(__file__).resolve().parents[3] / "alembic.ini"
|
||||
|
||||
|
||||
def _alembic_stamp_head() -> None:
|
||||
"""将当前 DB 标记为已到最新版本（基线既有 DB，不执行 SQL）"""
|
||||
cfg = Config(str(_ALEMBIC_INI))
|
||||
command.stamp(cfg, "head")
|
||||
|
||||
|
||||
def _alembic_upgrade_head() -> None:
|
||||
"""执行 alembic 迁移到最新版本"""
|
||||
cfg = Config(str(_ALEMBIC_INI))
|
||||
command.upgrade(cfg, "head")
|
||||
|
||||
|
||||
async def _run_alembic_migrations() -> None:
|
||||
"""以 alembic 管理 schema：既有未纳入管理的 DB 自动 stamp 基线，再 upgrade head
|
||||
|
||||
- 全新 DB：upgrade head 执行初始迁移，创建全部表
|
||||
- 既有已纳入管理：upgrade head 为 no-op
|
||||
- 既有但无 alembic_version（历史 DB）：先 stamp head 基线，再 upgrade（no-op）
|
||||
"""
|
||||
async with db_manager.engine.begin() as conn:
|
||||
has_alembic = await conn.execute(text("SELECT to_regclass('public.alembic_version')")).scalar()
|
||||
if not has_alembic:
|
||||
table_count = await conn.execute(
|
||||
text("SELECT count(*) FROM information_schema.tables WHERE table_schema='public' AND table_name <> 'alembic_version'")
|
||||
).scalar()
|
||||
if table_count and table_count > 0:
|
||||
logger.info("检测到既有 DB 未纳入 alembic 管理，自动 stamp head 作为基线")
|
||||
await asyncio.to_thread(_alembic_stamp_head)
|
||||
await asyncio.to_thread(_alembic_upgrade_head)
|
||||
|
||||
DEFAULT_PERMISSIONS = [
|
||||
{"code": "view_dashboard", "name": "查看仪表盘", "module": "dashboard"},
|
||||
{"code": "view_moldinsight", "name": "使用模具分析", "module": "moldinsight"},
|
||||
@@ -115,8 +150,7 @@ async def init_database(keep_connected: bool = True):
|
||||
"""初始化数据库"""
|
||||
try:
|
||||
await db_manager.connect()
|
||||
await db_manager.create_tables()
|
||||
await ensure_schema_updates()
|
||||
await _run_alembic_migrations()
|
||||
|
||||
async with db_manager.session() as session:
|
||||
perm_map = await init_permissions(session)
|
||||
@@ -153,97 +187,5 @@ async def init_database(keep_connected: bool = True):
|
||||
await db_manager.disconnect()
|
||||
|
||||
|
||||
async def ensure_schema_updates():
|
||||
async with db_manager.engine.begin() as conn:
|
||||
await conn.execute(text("ALTER TABLE products ADD COLUMN IF NOT EXISTS item_type VARCHAR(20) DEFAULT 'finished'"))
|
||||
await conn.execute(text("UPDATE products SET item_type = 'finished' WHERE item_type IS NULL"))
|
||||
await conn.execute(text("ALTER TABLE sales_orders ADD COLUMN IF NOT EXISTS production_status VARCHAR(20) DEFAULT 'not_started'"))
|
||||
await conn.execute(text("ALTER TABLE sales_orders ADD COLUMN IF NOT EXISTS production_no VARCHAR(50)"))
|
||||
await conn.execute(text("ALTER TABLE sales_orders ADD COLUMN IF NOT EXISTS planned_material_cost DOUBLE PRECISION DEFAULT 0"))
|
||||
await conn.execute(text("ALTER TABLE sales_orders ADD COLUMN IF NOT EXISTS actual_material_cost DOUBLE PRECISION DEFAULT 0"))
|
||||
await conn.execute(text("ALTER TABLE purchase_orders ADD COLUMN IF NOT EXISTS received_date TIMESTAMP WITHOUT TIME ZONE"))
|
||||
await conn.execute(text("ALTER TABLE purchase_orders ADD COLUMN IF NOT EXISTS paid_date TIMESTAMP WITHOUT TIME ZONE"))
|
||||
await conn.execute(text("""
|
||||
CREATE TABLE IF NOT EXISTS product_materials (
|
||||
id SERIAL PRIMARY KEY,
|
||||
finished_product_id INTEGER NOT NULL REFERENCES products(id),
|
||||
material_product_id INTEGER NOT NULL REFERENCES products(id),
|
||||
quantity DOUBLE PRECISION NOT NULL,
|
||||
loss_rate DOUBLE PRECISION DEFAULT 0,
|
||||
created_at TIMESTAMP DEFAULT NOW(),
|
||||
updated_at TIMESTAMP DEFAULT NOW()
|
||||
)
|
||||
"""))
|
||||
await conn.execute(text("""
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS uq_product_material_unique
|
||||
ON product_materials (finished_product_id, material_product_id)
|
||||
"""))
|
||||
await conn.execute(text("ALTER TABLE mold_cavity_data ADD COLUMN IF NOT EXISTS best_scheme_id VARCHAR(64)"))
|
||||
await conn.execute(text("ALTER TABLE mold_cavity_data ADD COLUMN IF NOT EXISTS confidence_score DOUBLE PRECISION"))
|
||||
await conn.execute(text("ALTER TABLE mold_cavity_data ADD COLUMN IF NOT EXISTS is_fallback BOOLEAN"))
|
||||
await conn.execute(text("ALTER TABLE mold_cavity_data ADD COLUMN IF NOT EXISTS fallback_reason TEXT"))
|
||||
await conn.execute(text("""
|
||||
CREATE INDEX IF NOT EXISTS idx_mold_cavity_best_scheme_id
|
||||
ON mold_cavity_data (best_scheme_id)
|
||||
"""))
|
||||
await conn.execute(text("""
|
||||
CREATE INDEX IF NOT EXISTS idx_mold_cavity_is_fallback
|
||||
ON mold_cavity_data (is_fallback)
|
||||
"""))
|
||||
await conn.execute(text("""
|
||||
DO $$
|
||||
BEGIN
|
||||
IF NOT EXISTS (
|
||||
SELECT 1 FROM pg_constraint
|
||||
WHERE conname = 'uq_inventory_product_warehouse'
|
||||
) THEN
|
||||
ALTER TABLE inventory
|
||||
ADD CONSTRAINT uq_inventory_product_warehouse UNIQUE (product_id, warehouse_id);
|
||||
END IF;
|
||||
END $$;
|
||||
"""))
|
||||
await conn.execute(text("""
|
||||
DO $$
|
||||
BEGIN
|
||||
IF NOT EXISTS (
|
||||
SELECT 1 FROM pg_constraint
|
||||
WHERE conname = 'ck_inventory_qty_nonnegative'
|
||||
) THEN
|
||||
ALTER TABLE inventory
|
||||
ADD CONSTRAINT ck_inventory_qty_nonnegative
|
||||
CHECK (quantity >= 0 AND locked_quantity >= 0 AND locked_quantity <= quantity);
|
||||
END IF;
|
||||
END $$;
|
||||
"""))
|
||||
await conn.execute(text("""
|
||||
DO $$
|
||||
BEGIN
|
||||
IF NOT EXISTS (
|
||||
SELECT 1 FROM pg_constraint
|
||||
WHERE conname = 'ck_purchase_order_items_qty'
|
||||
) THEN
|
||||
ALTER TABLE purchase_order_items
|
||||
ADD CONSTRAINT ck_purchase_order_items_qty
|
||||
CHECK (quantity > 0 AND received_quantity >= 0 AND received_quantity <= quantity);
|
||||
END IF;
|
||||
END $$;
|
||||
"""))
|
||||
await conn.execute(text("""
|
||||
DO $$
|
||||
BEGIN
|
||||
IF NOT EXISTS (
|
||||
SELECT 1 FROM pg_constraint
|
||||
WHERE conname = 'ck_sales_order_items_qty'
|
||||
) THEN
|
||||
ALTER TABLE sales_order_items
|
||||
ADD CONSTRAINT ck_sales_order_items_qty
|
||||
CHECK (quantity > 0 AND delivered_quantity >= 0 AND delivered_quantity <= quantity);
|
||||
END IF;
|
||||
END $$;
|
||||
"""))
|
||||
await conn.execute(text("ALTER TABLE sales_orders ADD COLUMN IF NOT EXISTS manufacturing_date TIMESTAMP WITHOUT TIME ZONE"))
|
||||
await conn.execute(text("ALTER TABLE sales_orders ALTER COLUMN manufacturing_date TYPE TIMESTAMP WITHOUT TIME ZONE"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(init_database(keep_connected=False))
|
||||
|
||||
@@ -1,55 +0,0 @@
|
||||
"""数据库迁移脚本 - 删除旧表并重新创建"""
|
||||
import asyncio
|
||||
from shared.database.database import db_manager
|
||||
from shared.models.database import Base
|
||||
from shared.utils.logger import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
async def migrate_database():
|
||||
"""迁移数据库：删除所有表并重新创建"""
|
||||
try:
|
||||
# 连接数据库
|
||||
await db_manager.connect()
|
||||
|
||||
# 删除所有表
|
||||
logger.info("正在删除所有数据库表...")
|
||||
async with db_manager.engine.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.drop_all)
|
||||
|
||||
# 重新创建所有表
|
||||
logger.info("正在创建所有数据库表...")
|
||||
async with db_manager.engine.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
|
||||
logger.info("数据库迁移完成！")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"数据库迁移失败: {e}")
|
||||
return False
|
||||
finally:
|
||||
await db_manager.disconnect()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import sys
|
||||
|
||||
# 检查命令行参数
|
||||
if len(sys.argv) > 1 and sys.argv[1] == '--force':
|
||||
confirm = 'yes'
|
||||
else:
|
||||
print("=== 数据库迁移 ===")
|
||||
print("警告：这将删除所有数据库表和数据！")
|
||||
confirm = input("确认继续？(yes/no): ")
|
||||
|
||||
if confirm.lower() == 'yes':
|
||||
asyncio.run(migrate_database())
|
||||
else:
|
||||
print("已取消迁移")
|
||||
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user