Alembic 深度指南：数据库迁移 + 自动生成 + 分支合并

数据库 Schema 变更管理是团队协作最容易出事故的一环：手动 DDL 改表忘记录入、生产与本地不一致、多分支合并后 migration 执行死锁。Alembic 作为 SQLAlchemy 官方迁移工具，能自动检测模型变更、版本化、可回滚、支持复杂合并。

自动生成与命名约定

命名约定（NAMING_CONVENTION）是多人协作的地基。不加命名约定的话，同一个约束在两台机器 autogenerate 名字随机（如约束名含随机hash），PR 一合并立刻 migration 冲突。env.py 配置 compare_type、compare_server_default 让自动检测更准，避免漏掉 bool 默认值变更、VARCHAR 长度变化这种常见的自动生成漏检。

# ====== alembic.ini 核心调优 ======
[alembic]
script_location = migrations
sqlalchemy.url = postgresql+psycopg://app:${DB_PASSWORD}@${DB_HOST}:5432/app
prepend_sys_path = .
timezone = Asia/Shanghai

[loggers]
keys = root,sqlalchemy,alembic
[handlers]
keys = console
[formatters]
keys = generic

[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S

# ====== migrations/env.py 支持自动生成 + 命名约定 ======
from __future__ import with_statement
from logging.config import fileConfig
from alembic import context
from sqlalchemy import engine_from_config, pool, MetaData
import os, sys

sys.path.insert(0, os.path.realpath(os.path.join(os.path.dirname(__file__), '..')))
from app.models import Base
from app.db.naming import NAMING_CONVENTION

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

target_metadata: MetaData = Base.metadata
target_metadata.naming_convention = NAMING_CONVENTION  # 关键：约束自动命名

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"},
        render_as_batch=True,  # SQLite 等支持批处理ALTER
        compare_type=True, compare_server_default=True,
        include_object=include_object_fn
    )
    with context.begin_transaction():
        context.run_migrations()

def include_object_fn(object, name, type_, reflected, compare_to):
    if type_ == "table" and name in {"spatial_ref_sys", "alembic_version"}:
        return False
    return True

def run_migrations_online() -> None:
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.", poolclass=pool.NullPool, future=True
    )
    with connectable.connect() as connection:
        context.configure(
            connection=connection, target_metadata=target_metadata,
            render_as_batch=True,
            compare_type=True,
            compare_server_default=True,
            include_object=include_object_fn,
            transaction_per_migration=True,  # 每个迁移独立事务
            lock_timeout=30                   # DDL 锁超时 30s
        )
        with context.begin_transaction():
            context.run_migrations()

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

# ====== app/db/naming.py — 约束命名公约防分支冲突 ======
NAMING_CONVENTION = {
    "ix": "ix_%(column_0_label)s",
    "uq": "uq_%(table_name)s_%(column_0_N_name)s",
    "ck": "ck_%(table_name)s_%(constraint_name)s",
    "fk": "fk_%(table_name)s_%(column_0_N_name)s_%(referred_table_name)s",
    "pk": "pk_%(table_name)s",
}

# ====== 典型迁移 revision 示例：data-only 迁移 ======
'''${message}

Revision ID: 0004_add_user_status
Revises: 0003_user_baseline
Create Date: 2024-06-15 11:22:33

'''
from alembic import op
import sqlalchemy as sa
from sqlalchemy import update
from sqlalchemy.orm import Session

# revision identifiers, used by Alembic.
revision = '0004_add_user_status'
down_revision = '0003_user_baseline'
branch_labels = None
depends_on = None

def upgrade() -> None:
    op.add_column('users', sa.Column('status', sa.String(16), nullable=False,
                  server_default='ACTIVE'))
    op.create_index('ix_users_status_created', 'users', ['status', 'created_at'])
    bind = op.get_bind(); sess = Session(bind)
    from app.models import User
    sess.execute(update(User).where(User.email.endswith('@test.com')).
                 values(status='INACTIVE'))
    sess.commit()

def downgrade() -> None:
    op.drop_index('ix_users_status_created', 'users')
    op.drop_column('users', 'status')

# ====== 多分支 HEAD 合并：建一个 merge migration ======
# 场景: Alice 分支 head=0005_add_posts, Bob 分支 head=0006_add_tags
# 解决:
#   alembic merge -m "merge posts and tags branches" 0005_add_posts 0006_add_tags
# 生成:
revision = '0007_merge_posts_tags'
down_revision = ('0005_add_posts', '0006_add_tags')
branch_labels = None
depends_on = None

def upgrade(): pass   # merge 迁移通常空实现，只解决 DAG 头分叉
def downgrade(): pass

分支合并与生产发布流程

Alice 和 Bob 同时拉两个 feature 分支，各写各的 migration，最后 merge 回 main 就会出现"两个 head"（分叉）。标准解法：alembic merge <hash1> <hash2> -m "merge A and B" 生成一个空的 merge revision，把两条链合成一条，down_revision 写成元组。生产发布：(1) 先备份数据库；(2) 事务+DDL锁超时；(3) 超大表（>100万行）用 CREATE INDEX CONCURRENTLY，禁止单一事务包裹长迁移。

最佳实践

每个迁移拆两类：Schema-only（结构变更，可快速回滚）和 Data-only（数据清洗回填，慎用）。禁止同一 migration 既改表又回写数据。每次 PR 发 CI 检查三件事：upgrade 跑一遍、downgrade 回前一版再 upgrade 一遍（幂等）、和主分支 head 数量一致。