FastAPI与Alembic数据库迁移完全指南

📂 所属阶段：第三阶段 — 数据持久化（数据库篇）
🔗 相关章节：FastAPI SQLAlchemy 2.0实战 · FastAPI依赖注入系统

为什么需要数据库迁移工具？

在没有迁移工具时，我们往往靠手写SQL管理表结构，这会带来一系列问题：

场景	传统做法	核心问题
添加字段	手动执行`ALTER TABLE`	难以回滚，生产风险高
修改表结构	直接改库	无版本控制，团队协作混乱
多环境同步	复制粘贴SQL	环境不一致，遗漏变更

Alembic的核心价值

作为SQLAlchemy官方迁移工具，它提供：

✅ Git式的数据库版本控制
✅ 自动对比模型与数据库生成脚本
✅ 可逆迁移，支持安全回滚
✅ 多环境结构一致性保证
✅ 完整的变更审计记录

项目依赖安装

# requirements.txt
fastapi==0.104.1
sqlalchemy==2.0.23
alembic==1.13.1
asyncpg==0.29.0  # PostgreSQL异步驱动
pydantic==2.5.0
uvicorn==0.24.0

安装命令：pip install -r requirements.txt

Alembic基础概念与项目集成

核心组件

env.py：连接数据库与SQLAlchemy模型的桥梁
alembic.ini：全局配置文件
versions/：存放版本化的迁移脚本
script.py.mako：迁移脚本生成模板

FastAPI项目结构

my_fastapi_project/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── models/
│   │   ├── __init__.py
│   │   └── base.py  # SQLAlchemy Base
│   └── database.py  # DATABASE_URL配置
├── alembic/
│   ├── env.py
│   ├── script.py.mako
│   └── versions/
├── alembic.ini
└── requirements.txt

初始化Alembic

# 在项目根目录执行
alembic init alembic

配置`env.py`（关键步骤）

修改alembic/env.py，引入项目的SQLAlchemy Base和数据库URL：

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

# 引入项目配置
from app.models.base import Base
from app.database import DATABASE_URL

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

if config.config_file_name:
    fileConfig(config.config_file_name)

target_metadata = Base.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 = 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,  # 对比列类型变更
            compare_server_default=True,  # 对比默认值
        )
        with context.begin_transaction():
            context.run_migrations()


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

简化`alembic.ini`

保留核心配置，删除冗余注释：

[alembic]
script_location = alembic
prepend_sys_path = .
file_template = %%(rev)s_%%(slug)s

[loggers]
keys = root,sqlalchemy,alembic

[handlers]
keys = console

[formatters]
keys = generic

[logger_root]
level = WARN
handlers = console

[logger_sqlalchemy]
level = WARN
qualname = sqlalchemy.engine

[logger_alembic]
level = INFO
qualname = alembic

[handler_console]
class = StreamHandler
args = (sys.stdout,)
level = NOTSET
formatter = generic

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

自动生成迁移脚本

基础命令

# 对比模型与当前数据库，自动生成迁移脚本
alembic revision --autogenerate -m "add user avatar column"

工作原理

扫描项目中的SQLAlchemy模型
连接数据库获取现有表结构
分析差异（新增/删除表、列、约束等）
生成包含upgrade和downgrade的脚本

生成的脚本示例

# alembic/versions/abc123_add_user_avatar_column.py
"""add user avatar column

Revision ID: abc123
Revises: def456
Create Date: 2026-03-26 14:00:00
"""
from alembic import op
import sqlalchemy as sa

revision = 'abc123'
down_revision = 'def456'
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.add_column('users', sa.Column('avatar', sa.String(500), nullable=True))


def downgrade() -> None:
    op.drop_column('users', 'avatar')

手动编写迁移脚本

何时需要手动编写？

需要同时迁移现有数据
自动生成无法处理的复杂SQL
需要条件判断或性能优化

数据迁移示例：字段类型转换

# 将users表status从字符串改为整数枚举
def upgrade() -> None:
    # 1. 添加临时列
    op.add_column('users', sa.Column('status_new', sa.Integer(), nullable=True))
    
    # 2. 迁移数据
    conn = op.get_bind()
    conn.execute(sa.text("""
        UPDATE users 
        SET status_new = CASE 
            WHEN status = 'active' THEN 1 ELSE 0
        END
    """))
    
    # 3. 替换列
    op.drop_column('users', 'status')
    op.alter_column('users', 'status_new', new_column_name='status', nullable=False)


def downgrade() -> None:
    op.add_column('users', sa.Column('status_old', sa.String(), nullable=True))
    conn = op.get_bind()
    conn.execute(sa.text("""
        UPDATE users 
        SET status_old = CASE WHEN status = 1 THEN 'active' ELSE 'inactive' END
    """))
    op.drop_column('users', 'status')
    op.alter_column('users', 'status_old', new_column_name='status', nullable=False)

版本控制与迁移命令

迁移文件依赖关系

每个迁移文件的down_revision必须指向前一个版本：

base → 001_initial → 002_add_posts → 003_add_comments → head

常用迁移命令

# 查看当前状态
alembic current
alembic history

# 升级/回滚
alembic upgrade head  # 升级到最新
alembic upgrade +1    # 升一级
alembic downgrade -1  # 回滚一级
alembic downgrade base  # 回滚到初始

# 生成SQL但不执行（用于审核）
alembic upgrade head --sql

多环境与生产环境部署

多环境配置

修改app/database.py，支持不同环境：

import os

ENV = os.getenv("APP_ENV", "development")
DATABASE_URLS = {
    "development": "postgresql+asyncpg://localhost:5432/myapp_dev",
    "testing": "postgresql+asyncpg://localhost:5432/myapp_test",
    "production": os.getenv("DATABASE_URL"),
}
DATABASE_URL = DATABASE_URLS[ENV]

环境特定命令

# 开发环境
APP_ENV=development alembic upgrade head

# 生产环境（使用环境变量数据库URL）
APP_ENV=production alembic upgrade head

Docker部署

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["sh", "-c", "alembic upgrade head && uvicorn app.main:app --host 0.0.0.0 --port 8000"]

安全的生产迁移流程

#!/bin/bash
set -e
echo "=== 开始生产迁移 ==="
# 1. 备份数据库
pg_dump -h $DB_HOST -U $DB_USER -d $DB_NAME > backup_$(date +%Y%m%d).sql
# 2. 生成SQL审核
alembic upgrade head --sql > migration.sql
cat migration.sql
read -p "确认执行？(yes/no): " confirm
if [[ $confirm != "yes" ]]; then exit 1; fi
# 3. 执行迁移
alembic upgrade head
echo "=== 迁移完成 ==="

常见陷阱与最佳实践

常见陷阱

❌ downgrade函数不完整：删除表前先删索引/约束
❌ 模型与数据库不一致：先确保当前库是head再改模型
❌ 大表逐行更新：用批量SQL代替

最佳实践

迁移文件组织：用序号或日期前缀命名，方便排序
完整的upgrade/downgrade：确保两者完全对称
测试迁移：在测试环境先验证迁移-回滚循环
生产前备份：永远在执行迁移前备份数据库

总结

Alembic为FastAPI+SQLAlchemy项目提供了完善的数据库版本控制方案，通过自动化脚本、可逆迁移和多环境支持，大幅降低了数据库结构变更的风险。

💡 关键要点：

始终在测试环境验证迁移

确保downgrade与upgrade完全对称

生产环境执行前必须备份

🔗 扩展阅读

#FastAPI与Alembic数据库迁移完全指南

#目录

#为什么需要数据库迁移工具？

#Alembic的核心价值

#项目依赖安装

#Alembic基础概念与项目集成

#核心组件

#FastAPI项目结构

#初始化Alembic

#配置env.py（关键步骤）

#简化alembic.ini

#自动生成迁移脚本

#基础命令

#工作原理

#生成的脚本示例

#手动编写迁移脚本

#何时需要手动编写？

#数据迁移示例：字段类型转换

#版本控制与迁移命令

#迁移文件依赖关系

#常用迁移命令

#多环境与生产环境部署

#多环境配置

#环境特定命令

#Docker部署

#安全的生产迁移流程

#常见陷阱与最佳实践

#常见陷阱

#最佳实践

#推荐迁移模板

#总结