Skip to content

37|Alembic 数据库迁移

开发过程中模型定义会不断变化:新增字段、修改类型、添加索引。直接改 Base.metadata.create_all() 只能创建新表,无法安全地修改已有表结构。Alembic 是 SQLAlchemy 官方的数据库迁移工具,它记录每一次结构变更(升级脚本和降级脚本),让数据库结构的演进可追踪、可回滚。

一、安装与初始化

bash
uv add alembic
bash
uv run alembic init alembic

执行后生成:

text
alembic/
├── env.py          # 迁移环境配置
├── script.py.mako  # 迁移脚本模板
├── versions/       # 存放生成的迁移脚本(空)
└── alembic.ini     # 配置文件

修改 alembic.ini 中的数据库连接:

ini
sqlalchemy.url = sqlite:///ops.db

修改 alembic/env.py 导入模型基类:

python
from database import Base   # 你的 Base 定义

target_metadata = Base.metadata

二、生成迁移脚本

模型定义修改后,生成迁移脚本:

bash
uv run alembic revision --autogenerate -m "create servers table"

Alembic 对比当前数据库结构和 Base.metadata,自动生成差异脚本,保存在 alembic/versions/ 下:

python
# alembic/versions/xxxx_create_servers_table.py
from alembic import op
import sqlalchemy as sa

revision = "abc123"
down_revision = None
branch_labels = None
depends_on = None

def upgrade():
    op.create_table(
        "servers",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("hostname", sa.String(length=64), nullable=False),
        sa.PrimaryKeyConstraint("id"),
    )

def downgrade():
    op.drop_table("servers")
函数作用
upgrade()升级:应用变更
downgrade()降级:回滚变更

autogenerate 不是万能的,复杂的变更(如列类型修改、数据迁移)需要手动编辑脚本。

三、执行迁移

升级到最新版本

bash
uv run alembic upgrade head

降级一个版本

bash
uv run alembic downgrade -1

降级到指定版本

bash
uv run alembic downgrade abc123

查看当前版本

bash
uv run alembic current

查看历史

bash
uv run alembic history

四、常见变更操作

添加列

python
def upgrade():
    op.add_column("servers", sa.Column("env", sa.String(20), nullable=True))

def downgrade():
    op.drop_column("servers", "env")

添加索引

python
def upgrade():
    op.create_index("idx_servers_status", "servers", ["status"])

def downgrade():
    op.drop_index("idx_servers_status", table_name="servers")

修改列类型(SQLite 有限制)

SQLite 的 ALTER TABLE 不支持直接修改列类型,需要重建表:

python
def upgrade():
    # Alembic 的 batch_alter_table 自动处理 SQLite 限制
    with op.batch_alter_table("servers") as batch_op:
        batch_op.alter_column("port", type_=sa.String(10))

数据迁移

结构变更同时需要迁移数据时,在 upgrade() 中写 Python 逻辑:

python
def upgrade():
    # 1. 添加新列
    op.add_column("servers", sa.Column("env", sa.String(20), nullable=True))

    # 2. 填充默认值
    op.execute("UPDATE servers SET env = 'dev' WHERE env IS NULL")

    # 3. 改为非空
    with op.batch_alter_table("servers") as batch_op:
        batch_op.alter_column("env", nullable=False)

五、团队协作规范

迁移脚本必须可重复执行

不要在 upgrade() 中写 INSERT 固定数据(除非是种子数据迁移),因为同事的数据库可能已经存在这些数据:

python
# 错误:重复执行会报错
def upgrade():
    op.execute("INSERT INTO projects (name) VALUES ('default')")

# 正确:用 INSERT OR IGNORE / ON CONFLICT
def upgrade():
    op.execute("INSERT OR IGNORE INTO projects (id, name) VALUES (1, 'default')")

提交迁移脚本到版本控制

alembic/versions/ 下的脚本必须和代码一起提交到 Git。新同事拉取代码后执行 alembic upgrade head 即可同步数据库结构。

生产环境迁移

bash
# 1. 先备份数据库
# 2. 在 staging 环境验证迁移脚本
# 3. 生产环境执行
alembic upgrade head

不要在生产环境直接用 --autogenerate,先在开发环境生成脚本、人工审查、测试通过后再应用到生产。

六、常见错误

模型改了但 migration 没生成

bash
# 错误:直接改数据库,没有生成脚本
# 正确:改模型 → 生成脚本 → 执行脚本
uv run alembic revision --autogenerate -m "add column"
uv run alembic upgrade head

downgrade 未实现

python
# 错误:downgrade 为空,回滚时失败
def downgrade():
    pass

# 正确:每个 upgrade 操作都对应 downgrade

多人同时生成迁移脚本导致冲突

两个开发者同时基于同一版本生成脚本,提交后会产生分支。解决方案:

bash
# 合并分支
uv run alembic merge heads -m "merge branches"

更推荐的做法:每次开发新功能前先 git pull,确保基于最新的 migration 生成脚本。