⚡

数据库迁移Skill db-migration

数据库迁移技能提供专业的Alembic迁移管理解决方案，专为SQLModel/FastAPI项目设计。该技能涵盖数据库模式变更的完整生命周期管理，包括迁移生成、版本控制、安全升级和回滚操作。核心功能包括自动迁移生成、数据迁移处理、批量操作优化和迁移质量保证。适用于后端开发、DevOps和数据库管理场景，确保数据库结构演进的安全性和可靠性。关键词：Alembic迁移、数据库模式管理、SQLModel集成、FastAPI数据库、版本控制、数据迁移、回滚策略、DevOps数据库

后端开发 0 次安装 0 次浏览 更新于 3/2/2026

name: db-migration description: | 用于设置Alembic迁移或进行数据库模式更改时使用。 触发场景：初始化Alembic、生成迁移、应用升级、回滚更改或创建数据迁移。 不适用于：迁移上下文之外的原始SQL执行或非数据库模式更新。

数据库迁移技能

为SQLModel/FastAPI项目提供专业的Alembic迁移管理，支持安全的模式演进和回滚能力。

快速参考

命令 用途
alembic init alembic 在项目中初始化Alembic
alembic revision --autogenerate -m "message" 根据模型变更生成迁移
alembic revision -m "message" 手动创建空迁移
alembic upgrade head 应用所有待处理的迁移
alembic upgrade +1 一次应用一个迁移
alembic downgrade -1 回滚最后一个迁移
alembic downgrade base 回滚所有迁移
alembic current 显示当前版本
alembic history 显示迁移历史

初始设置

1. 初始化Alembic

alembic init alembic

2. 配置alembic.ini

# alembic.ini
sqlalchemy.url = driver://user:pass@localhost/dbname
file_template = %%(year)s_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d_%%(rev)s_%%(slug)s
timezone = UTC

3. 为SQLModel配置env.py

# alembic/env.py
from logging.config import fileConfig
from sqlalchemy import pool
from sqlalchemy.engine import Connection
from alembic.runtime.migration import MigrationContext
from sqlmodel import SQLModel, create_engine
from myapp.models import *  # 导入所有SQLModel类

config = context.config
config.set_main_option("sqlalchemy.url", "postgresql://user:pass@localhost/dbname")

target_metadata = SQLModel.metadata


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 = create_engine(
        config.get_main_option("sqlalchemy.url"),
        poolclass=pool.NullPool,
    )
    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
        )
        with context.begin_transaction():
            context.run_migrations()


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

生成迁移

根据模型变更自动生成

# 根据模型差异自动生成迁移
alembic revision --autogenerate -m "add_fees_table"

# 使用特定版本范围
alembic revision --autogenerate -m "add_user_email" --rev-id=abc123

手动迁移

# 为手动更改创建空迁移
alembic revision -m "add_status_column"

示例：添加新表

# alembic/versions/2024_01_15_1200_add_fees_table.py
"""add_fees_table

Revision ID: abc123
Revises: def456
Create Date: 2024-01-15 12:00:00.000000

"""
from alembic import op
import sqlalchemy as sa
from sqlmodel import SQLModel

# 版本标识符
revision = 'abc123'
down_revision = 'def456'
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.create_table(
        'fees',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('student_id', sa.Integer(), nullable=False),
        sa.Column('amount', sa.Numeric(precision=10, scale=2), nullable=False),
        sa.Column('status', sa.String(length=20), nullable=False, default='pending'),
        sa.Column('due_date', sa.DateTime(), nullable=False),
        sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.func.now()),
        sa.Column('updated_at', sa.DateTime(), nullable=False, server_default=sa.func.now()),
        sa.PrimaryKeyConstraint('id'),
        sa.ForeignKeyConstraint(['student_id'], ['students.id']),
    )
    op.create_index('ix_fees_student_id', 'fees', ['student_id'])
    op.create_index('ix_fees_status', 'fees', ['status'])


def downgrade() -> None:
    op.drop_index('ix_fees_status', table_name='fees')
    op.drop_index('ix_fees_student_id', table_name='fees')
    op.drop_table('fees')

示例：添加列

# alembic/versions/2024_01_16_0900_add_fees_description.py
"""add_fees_description

Revision ID: ghi789
Revises: abc123
Create Date: 2024-01-16 09:00:00.000000

"""
from alembic import op


def upgrade() -> None:
    op.add_column('fees', sa.Column('description', sa.Text(), nullable=True))


def downgrade() -> None:
    op.drop_column('fees', 'description')

应用迁移

标准升级

# 升级到最新版本
alembic upgrade head

# 一次升级一步
alembic upgrade +1

# 升级到特定版本
alembic upgrade abc123

试运行（检查将发生什么）

# 显示待处理的迁移而不应用
alembic show heads
alembic history --verbose

回滚（降级）

# 回滚一个迁移
alembic downgrade -1

# 回滚到特定版本
alembic downgrade abc123

# 回滚所有迁移（清空数据库）
alembic downgrade base

安全降级模式

def downgrade() -> None:
    # 始终在表之前删除索引
    op.drop_index('ix_fees_status', table_name='fees')
    op.drop_index('ix_fees_student_id', table_name='fees')
    # 在表之前删除外键
    op.drop_constraint('fees_student_id_fkey', 'fees', type_='foreignkey')
    op.drop_table('fees')

数据迁移

示例：批量更新的数据迁移

# alembic/versions/2024_01_17_1400_update_fees_status.py
"""update_fees_status_values

Revision ID: jkl012
Revises: ghi789
Create Date: 2024-01-17 14:00:00.000000

"""
from alembic import op
from sqlalchemy import text


def upgrade() -> None:
    # 更新现有记录
    op.execute(
        text("UPDATE fees SET status = 'pending' WHERE status = 'unpaid'")
    )


def downgrade() -> None:
    # 恢复状态值
    op.execute(
        text("UPDATE fees SET status = 'unpaid' WHERE status = 'pending'")
    )

示例：枚举迁移

def upgrade() -> None:
    # 添加新的枚举类型
    op.execute("CREATE TYPE fee_status_new AS ENUM ('pending', 'paid', 'overdue', 'waived')")
    # 将数据复制到新类型
    op.execute("ALTER TABLE fees ALTER COLUMN status TYPE fee_status_new USING status::text::fee_status_new")
    # 删除旧类型
    op.execute("DROP TYPE fee_status_old")


def downgrade() -> None:
    # 反转过程
    op.execute("ALTER TABLE fees ALTER COLUMN status TYPE VARCHAR(20)")
    op.execute("DROP TYPE fee_status_new")

质量检查清单

  • [ ] 数据迁移：修改列/表时处理现有数据
  • [ ] 测试迁移：在测试环境中运行alembic upgrade然后alembic downgrade
  • [ ] 幂等操作：up()和down()可以安全地多次运行
  • [ ] 无数据丢失：使用DROP TABLE IF EXISTS、DROP COLUMN IF EXISTS
  • [ ] 创建索引：在升级中包含索引创建，在降级中删除
  • [ ] 外键：处理约束顺序（先创建，后删除）
  • [ ] 向后兼容：迁移期间不破坏现有应用程序

与其他技能的集成

技能 集成点
@sqlmodel-crud 模型变更触发迁移
@fastapi-app 迁移在启动时或通过CLI运行
@jwt-auth 可能需要在迁移期间处理身份验证

迁移最佳实践

1. 始终在手动编辑前生成

alembic revision --autogenerate -m "describe_change"
# 然后审查并编辑生成的文件

2. 审查生成的迁移

# 检查：
# - 列类型是否匹配SQLModel定义
# - 外键约束是否正确
# - 索引是否合适
# - 是否设置了默认值

3. 测试迁移周期

# 在测试环境中
alembic downgrade base
alembic upgrade head

# 验证所有数据是否完整

4. 处理长时间运行的迁移

# 对于大表，使用批量更新
def upgrade():
    op.execute("""
        UPDATE fees SET status = 'pending'
        WHERE status IS NULL
        LIMIT 10000
    """)

目录结构

project/
├── alembic/
│   ├── env.py              # 迁移配置
│   ├── script.py.mako      # 新迁移模板
│   ├── README              # Alembic文档
│   └── versions/
│       ├── 2024_01_15_1200_add_fees_table.py
│       └── 2024_01_16_0900_add_fees_description.py
├── myapp/
│   └── models.py           # SQLModel定义
└── alembic.ini             # Alembic配置