Alembic 迁移与种子数据

第 11 章 · Alembic 迁移与种子数据

本章目标：在 db-demo 中 init Alembic；配置 env.py 绑定 ORM Base.metadata；使用 revision --autogenerate 生成迁移；掌握 upgrade / downgrade 与版本链；编写 种子数据 脚本；对照 fastapi-web ch06 的异步 Alembic 与 Repository 模式差异。

学时建议：5～6 小时（含 2 小时迁移跟练）

前置：完成 python-database ch10（ORM 模型齐全）；了解 ch07 DDL 基线。

11.1 场景说明：schema 必须可版本化

db-demo 演进示例：

1. v1：ch07 六表
2. v2：ch10 增加 tags / product_tags
3. v3：products 增加 cover_url（练习）

对照 fastapi-web ch06：svc-demo 用 异步 engine + Repository；db-demo 用 同步 Alembic（官方主流），原理相同。

11.2 安装与 init

cd ~/python-learn/db-demo
pip install "alembic>=1.13,<2"
pip freeze | grep -i alembic >> requirements.txt
alembic init alembic

生成结构：

db-demo/
├── alembic/
│   ├── versions/           # 迁移脚本目录
│   ├── env.py              # 运行时入口
│   ├── README
│   └── script.py.mako      # 新 revision 模板
├── alembic.ini
└── orm/

alembic.ini 关键项：

[alembic]
script_location = alembic
# sqlalchemy.url 可留空，由 env.py 注入
prepend_sys_path = .

11.3 配置 env.py（同步版）

alembic/env.py（db-demo 精简模板）：

from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context

from core.engine import DATABASE_URL
from orm.base import Base
# 必须 import 所有模型，否则 autogenerate 漏表
from orm.models import category, product, tag, order, cart, user  # noqa: F401

config = context.config
config.set_main_option("sqlalchemy.url", DATABASE_URL)
target_metadata = Base.metadata

if config.config_file_name is not None:
    fileConfig(config.config_file_name)


def run_migrations_offline() -> None:
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )
    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online() -> None:
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            compare_type=True,
            render_as_batch=True,    # SQLite ALTER 友好
        )
        with context.begin_transaction():
            context.run_migrations()


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

11.4 与 fastapi-web ch06 对照

fastapi-web ch06 异步 env 核心片段（对照阅读）：

async def run_async_migrations() -> None:
    connectable = async_engine_from_config(..., poolclass=pool.NullPool)
    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)
    await connectable.dispose()

db-demo 学员先掌握 同步 Alembic；转 async 仅改 env.py 与 engine 来源。

11.5 首次 autogenerate 与 upgrade

准备：备份或删除旧库，让 autogenerate 感知「空库 → 全模型」。

rm -f data/shop_db.sqlite3
alembic revision --autogenerate -m "init shop schema"
alembic upgrade head

检查 alembic/versions/xxxx_init_shop_schema.py：

def upgrade() -> None:
    op.create_table("categories", ...)
    op.create_table("products", ...)
    # ...

def downgrade() -> None:
    op.drop_table("products")
    op.drop_table("categories")
    # 顺序：先删子表

审查清单：

• [ ] upgrade / downgrade 成对可逆
• [ ] FK、unique、server_default 正确
• [ ] products.price 为 INTEGER（分）
• [ ] is_published 默认值合理
• [ ] 无多余 drop_table（会丢数据）

11.6 增量迁移示例：增加 cover_url

修改 Product 模型：

cover_url: Mapped[str | None] = mapped_column(String(512), nullable=True)

生成并应用：

alembic revision --autogenerate -m "add product cover_url"
# 人工编辑：确认仅为 add_column
alembic upgrade head

预期 upgrade()：

def upgrade() -> None:
    with op.batch_alter_table("products", schema=None) as batch_op:
        batch_op.add_column(sa.Column("cover_url", sa.String(length=512), nullable=True))

回滚：

alembic downgrade -1
alembic current
alembic history --verbose

11.7 字段演进迁移：sku → slug（教学脚本）

若从旧库升级（对照 django-web ch07），勿 drop 列，用 rename + 数据迁移：

"""evolve sku to slug

Revision ID: xxxx
"""
from alembic import op
import sqlalchemy as sa