Alembic Deep Dive: Migrations, Autogenerate, and Branch Merging

Schema change management is the most accident-prone part of team collaboration: handwritten DDL not tracked, prod/dev out of sync, deadlocks from migrations after multi-branch merge. As SQLAlchemy's official tool, Alembic auto-detects model changes, versions them rollbackably, and handles complex merges.

Autogenerate and Naming Convention

Naming convention (NAMING_CONVENTION) is the foundation of team collaboration. Without it, constraint names on two dev machines autogenerate randomly (hash suffixes), causing migration conflicts on PR merge the instant they land. env.py options compare_type + compare_server_default boost detection accuracy, catching common false-negatives like bool default or VARCHAR length tweaks.

# ====== 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

Branch Merging and Production Release Flow

Alice and Bob open two feature branches with separate migrations; merging back to main yields "two divergent heads". Standard fix: alembic merge <hash1> <hash2> -m "merge A and B" produces an empty merge revision tying both chains together, with down_revision as a tuple. Production playbook: (1) take DB backup first; (2) per-migration transactions with DDL lock timeout; (3) large tables (>1M rows) use CREATE INDEX CONCURRENTLY — never wrap long migrations in single txn.

Best Practices

Split each migration into Schema-only (structural change, rollback-safe) vs Data-only (backfill/cleanup - use sparingly). Never mix DDL and bulk writes in same migration. Every PR CI runs three checks: upgrade once, downgrade one revision then upgrade again (idempotency), count of migration heads matches main branch count.