AlembicMigrationPatternsSkill py-alembic-patterns

Alembic 迁移模式

问题陈述

Alembic 自动生成很方便，但有时会遗漏一些内容，有时还会生成危险的迁移。模式变更是高风险的——不良的迁移会导致数据丢失或停机。每次迁移都需要人工审核。

---

模式：迁移命令

# 从模型变更生成迁移
uv run alembic revision --autogenerate -m "添加用户偏好设置表"

# 应用迁移
uv run alembic upgrade head

# 回滚一个迁移
uv run alembic downgrade -1

# 回滚到特定版本
uv run alembic downgrade abc123

# 显示当前版本
uv run alembic current

# 显示迁移历史
uv run alembic history

# 显示待处理迁移
uv run alembic history --indicate-current

---

模式：审查自动生成的迁移

总是审查自动生成的迁移。它们经常需要修复。

自动生成能捕获的内容

• 表的创建/删除
• 列的添加/移除
• 列类型更改
• 外键更改
• 索引更改（有时）

自动生成遗漏的内容

• 列重命名（视为删除+添加=数据丢失）
• 表重命名（同样的问题）
• 数据迁移
• 约束名称
• 部分索引
• 复杂索引更改
• 检查约束
• 触发器和函数

# ❌ 危险：自动生成用于列重命名
def upgrade():
    op.drop_column("users", "name")      # 数据丢失！
    op.add_column("users", sa.Column("full_name", sa.String()))

# ✅ 正确：手动重命名
def upgrade():
    op.alter_column("users", "name", new_column_name="full_name")

def downgrade():
    op.alter_column("users", "full_name", new_column_name="name")

---

模式：安全的迁移结构

"""添加用户偏好设置表。

修订 ID: abc123
修订: def456
创建日期: 2024-01-15 10:30:00.000000
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql

# 修订标识符
revision = "abc123"
down_revision = "def456"
branch_labels = None
depends_on = None


def upgrade() -> None:
    # 总是明确，永远不要依赖默认值
    op.create_table(
        "user_preferences",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
        sa.Column("user_id", postgresql.UUID(as_uuid=True), nullable=False),
        sa.Column("theme", sa.String(50), nullable=False, server_default="light"),
        sa.Column("notifications_enabled", sa.Boolean(), nullable=False, server_default="true"),
        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
    )
    
    # 明确的索引名称
    op.create_index(
        "ix_user_preferences_user_id",
        "user_preferences",
        ["user_id"],
    )
    
    # 外键与明确的名称
    op.create_foreign_key(
        "fk_user_preferences_user_id",
        "user_preferences",
        "users",
        ["user_id"],
        ["id"],
        ondelete="CASCADE",
    )


def downgrade() -> None:
    # 总是实现降级！
    op.drop_constraint("fk_user_preferences_user_id", "user_preferences", type_="foreignkey")
    op.drop_index("ix_user_preferences_user_id", "user_preferences")
    op.drop_table("user_preferences")

---

模式：添加非空列

**问题：**向现有表添加 NOT NULL 列如果表中有行将会失败。

# ❌ 错误：如果表中有数据则失败
def upgrade():
    op.add_column("users", sa.Column("role", sa.String(50), nullable=False))

# ✅ 正确：三步过程
def upgrade():
    # 第一步：作为可空添加
    op.add_column("users", sa.Column("role", sa.String(50), nullable=True))
    
    # 第二步：回填现有行
    op.execute("UPDATE users SET role = 'member' WHERE role IS NULL")
    
    # 第三步：添加 NOT NULL 约束
    op.alter_column("users", "role", nullable=False)

def downgrade():
    op.drop_column("users", "role")

---

模式：数据迁移

**问题：**需要在模式变更期间转换现有数据。

from sqlalchemy import text

def upgrade():
    # 获取数据操作的连接
    connection = op.get_bind()
    
    # 添加新列
    op.add_column("assessments", sa.Column("status", sa.String(20)))
    
    # 迁移数据
    connection.execute(
        text("""
            UPDATE assessments 
            SET status = CASE 
                WHEN completed_at IS NOT NULL THEN 'completed'
                WHEN started_at IS NOT NULL THEN 'in_progress'
                ELSE 'pending'
            END
        """)
    )
    
    # 现在可以安全地添加 NOT NULL
    op.alter_column("assessments", "status", nullable=False)


def downgrade():
    op.drop_column("assessments", "status")

---

模式：大型表迁移

**问题：**大型表的迁移可能会锁定表太久。

def upgrade():
    # ✅ 正确：并发添加索引（无锁定）
    op.execute(
        "CREATE INDEX CONCURRENTLY ix_events_user_id ON events (user_id)"
    )
    
    # 注意：CONCURRENTLY 需要自动提交模式
    # 添加到迁移文件：
    # from alembic import context
    # context.configure(transaction_per_migration=False)

def downgrade():
    op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_events_user_id")


# 对于大型表的列变更，考虑：
# 1. 添加新列（可空）
# 2. 通过单独的脚本分批回填
# 3. 在单独的迁移中添加约束

---

模式：枚举变更

**问题：**PostgreSQL 枚举很难修改。

# 向现有枚举添加值
def upgrade():
    # PostgreSQL 特定：向枚举添加值
    op.execute("ALTER TYPE assessment_status ADD VALUE 'archived'")

def downgrade():
    # PostgreSQL 中不能移除枚举值！
    # 选项：
    # 1. 保留它（通常没问题）
    # 2. 重新创建枚举（复杂，需要数据迁移）
    pass


# 创建新枚举
def upgrade():
    # 首先创建枚举类型
    assessment_status = postgresql.ENUM(
        "draft", "active", "completed", "archived",
        name="assessment_status",
        create_type=True,
    )
    assessment_status.create(op.get_bind())
    
    # 然后使用它
    op.add_column(
        "assessments",
        sa.Column("status", assessment_status, nullable=False, server_default="draft"),
    )
def downgrade():
    op.drop_column("assessments", "status")
    op.execute("DROP TYPE assessment_status")

---

模式：多个头（分支）

**问题：**多个开发人员同时创建迁移。

# 检查多个头
uv run alembic heads

# 如果有多个头，创建合并迁移
uv run alembic merge -m "合并头" abc123 def456

# 或指定 down_revision 为元组
down_revision = ("abc123", "def456")

---

模式：测试迁移

# test_migrations.py
import pytest
from alembic import command
from alembic.config import Config

@pytest.fixture
def alembic_config():
    config = Config("alembic.ini")
    return config

def test_upgrade_downgrade(alembic_config, test_db):
    """测试迁移可以升级和降级。"""
    # 升级到头
    command.upgrade(alembic_config, "head")
    
    # 降级到基础
    command.downgrade(alembic_config, "base")
    
    # 再次升级
    command.upgrade(alembic_config, "head")

def test_migration_has_downgrade():
    """确保所有迁移都有降级。"""
    # 解析迁移文件并检查降级不是只是 'pass'
    ...

---

迁移审查清单

在应用任何迁移之前：

• [ ] 实现了降级函数（不仅仅是 pass）
• [ ] 列重命名使用 alter_column，而不是删除+添加
• [ ] 添加非空列时使用默认值或回填
• [ ] 大型表操作考虑锁定
• [ ] 索引有明确名称
• [ ] 外键有明确的名称和 ON DELETE 行为
• [ ] 枚举在使用前创建
• [ ] 数据迁移用实际数据量进行测试
• [ ] 迁移测试：升级，降级，升级

---

生产安全

# 设置语句超时以防止长时间锁定
def upgrade():
    op.execute("SET statement_timeout = '5s'")
    
    # 你的迁移在这里
    
    op.execute("SET statement_timeout = '0'")  # 重置

# 生产迁移前总是备份
pg_dump -h host -U user -d dbname > backup_before_migration.sql

# 使用 --sql 预览
uv run alembic upgrade head --sql

# 真正应用
uv run alembic upgrade head

---

常见问题

问题	可能的原因	解决方案
“目标数据库不更新”	待处理迁移	运行 alembic upgrade head
“无法定位版本”	缺少迁移文件	检查版本历史
多个头	并发开发	创建合并迁移
锁定超时	长时间运行的迁移	使用 CONCURRENTLY，批量更新
部署时数据丢失	列重命名为删除+添加	仔细审查自动生成