# Flask-Migrate 生产实践：从模型变更到数据库平滑迁移

在 Flask 项目里使用 ORM 时，很多人一开始会直接调用 db.create_all()，让 SQLAlchemy 根据模型把表创建出来。这个方式适合项目早期 demo，但它并不适合持续演进的生产系统。

原因很简单：db.create_all() 只负责“缺表时建表”，不会帮你修改已经存在的表。当 ORM 模型新增字段、修改字段类型、增加索引、拆表、合表时，它都不会自动把数据库结构同步过去。

更危险的做法是先 db.drop_all() 再 db.create_all()。表结构确实会变成新的，但原有数据也会被清空。生产项目里，数据库结构变更不是“重新建一下表”这么轻的事，而是一次需要评审、验证、发布和监控的工程动作。

Flask-Migrate 解决的就是这个问题。它基于 Alembic，为 Flask + Flask-SQLAlchemy 项目提供数据库迁移能力：把每一次结构变更记录成版本脚本，再通过命令把数据库从一个版本升级到另一个版本。

# 迁移工具到底解决什么

数据库迁移解决的是三个生产问题。

第一，结构变化要可追踪。

模型代码只能说明“现在应该长什么样”，但不能说明“从上一个版本怎么变到现在”。迁移脚本记录的是变化过程，例如新增字段、创建索引、修改约束、补充默认值等。

第二，多环境结构要一致。

开发、测试、预发、生产环境不能靠人工记忆去改表。迁移脚本进入 Git 后，每个环境都可以执行相同版本链路，数据库结构才有一致性基础。

第三，变更要能被审查和回退。

迁移脚本里有 upgrade() 和 downgrade()。前者描述升级，后者描述回滚。即使生产环境未必真的执行自动回滚，至少也能让团队在上线前明确“这次变更是否可逆、不可逆风险在哪里”。

# `db.create_all()` 的边界

假设一开始有一个用户模型：

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), nullable=False)

1
2
3

后来新增邮箱字段：

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), nullable=False)
    email = db.Column(db.String(128), unique=True)

1
2
3
4

如果数据库里已经存在 user 表，再执行 db.create_all()，不会自动多出 email 字段。SQLAlchemy 不会替你推断“应该 ALTER TABLE”。这是设计上的边界，不是 bug。

所以生产项目里通常遵循一个原则：

本地临时 demo 可以使用 db.create_all()。
正式项目初始化后，表结构统一交给迁移脚本管理。
一旦迁移目录建立，后续不要再依赖 create_all() 修改结构。

# 安装与初始化

安装扩展：

pip install flask-migrate

在项目里初始化 Migrate：

from flask_migrate import Migrate

migrate = Migrate()
migrate.init_app(app, db, directory="internal/migration")

1
2
3
4

几个关键点：

app 是 Flask 应用实例。
db 是 Flask-SQLAlchemy 的 SQLAlchemy 实例。
directory 是迁移文件目录，不传时默认是 migrations。

生产项目一般会采用应用工厂模式，把扩展对象集中放在一个模块中。

# internal/extensions.py
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate

db = SQLAlchemy()
migrate = Migrate()

1
2
3
4
5
6

# app/server.py
from flask import Flask
from internal.extensions import db, migrate

def create_app():
    app = Flask(__name__)
    app.config.from_object("config")

    db.init_app(app)
    migrate.init_app(app, db, directory="internal/migration")

    return app

app = create_app()

1
2
3
4
5
6
7
8
9
10
11
12
13
14

这里建议固定迁移目录，并把它提交到 Git。迁移目录不是构建产物，而是项目源代码的一部分。

# 基础命令

Flask-Migrate 会扩展 Flask CLI，提供 flask db 命令。假设应用入口是 app.server:app，命令可以这样写。

初始化迁移环境：

flask --app app.server:app db init

生成迁移脚本：

flask --app app.server:app db migrate -m "add email to user"

执行升级：

flask --app app.server:app db upgrade

回滚一个版本：

flask --app app.server:app db downgrade

回滚到最初版本：

flask --app app.server:app db downgrade base

回滚到指定版本：

flask --app app.server:app db downgrade <revision>

查看当前数据库版本：

flask --app app.server:app db current

查看迁移历史：

flask --app app.server:app db history

查看迁移头版本：

flask --app app.server:app db heads

生成 SQL，不直接执行：

flask --app app.server:app db upgrade --sql

--sql 在生产评审里很有用。它可以让你看到最终会执行哪些 SQL，再决定是否直接由应用发布流程执行，还是交给 DBA 或数据库发布平台执行。

# 一次标准的开发流程

开发环境可以按下面的流程走：

修改 ORM 模型。
执行 flask --app app.server:app db migrate -m "清晰的变更说明"。
打开生成的迁移脚本，检查 upgrade() 和 downgrade()。
执行 flask --app app.server:app db upgrade。
本地跑测试，确认接口、模型和数据库结构一致。
将模型代码和迁移脚本一起提交。

这里最容易犯的错是“只提交模型，不提交迁移脚本”。模型代码和迁移脚本必须成对出现，否则其他环境无法通过统一方式升级数据库。

# 迁移脚本必须人工检查

flask db migrate 会根据模型和数据库当前结构做自动比较，但自动生成不等于正确。

下面这些场景尤其要手动检查。

# 字段重命名

如果把 username 改成 name，Alembic 可能会识别成“删除 username，新增 name”，而不是“重命名字段”。

危险脚本可能长这样：

def upgrade():
    op.add_column("user", sa.Column("name", sa.String(length=64), nullable=True))
    op.drop_column("user", "username")

1
2
3

这会导致原来的 username 数据丢失。更稳的方式是明确写成重命名，或在不同数据库下写对应 SQL。

def upgrade():
    op.alter_column("user", "username", new_column_name="name")


def downgrade():
    op.alter_column("user", "name", new_column_name="username")

1
2
3
4
5
6

不同数据库对重命名字段的支持不完全一致，生产前要在目标数据库版本上验证。

# 新增非空字段

如果给已有大表新增 nullable=False 字段，没有默认值时可能直接失败，因为历史数据无法满足非空约束。

不要一上来就写：

op.add_column("user", sa.Column("status", sa.String(length=20), nullable=False))

更稳的三阶段做法是：

先新增可空字段。
后台补齐历史数据。
再把字段改成非空。

迁移脚本可以拆成两个版本，避免一次上线同时承担结构变更和大量数据修复。

def upgrade():
    op.add_column("user", sa.Column("status", sa.String(length=20), nullable=True))

1
2

补数完成后，再做第二次迁移：

def upgrade():
    op.alter_column("user", "status", nullable=False)

1
2

# 创建索引

索引能提升查询性能，但在大表上创建索引可能锁表或拖慢写入。

如果是 PostgreSQL，生产大表经常会考虑 CREATE INDEX CONCURRENTLY，减少锁表影响。但它不能在事务块内执行，需要在 Alembic 里特殊处理。

示例：

from alembic import op


def upgrade():
    with op.get_context().autocommit_block():
        op.execute("CREATE INDEX CONCURRENTLY ix_user_email ON \"user\" (email)")


def downgrade():
    with op.get_context().autocommit_block():
        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_user_email")

1
2
3
4
5
6
7
8
9
10
11

如果是 MySQL，也要关注当前存储引擎、版本、DDL 算法、锁等待和执行时间。不要把“创建索引”当成一个无成本操作。

# 删除字段和删除表

删除字段、删除表属于高风险操作。即使业务代码已经不再读取，也建议延迟删除。

更稳的流程是：

先发布代码，停止写入和读取旧字段。
观察一段时间，确认没有旧逻辑依赖。
备份或归档相关数据。
再单独发布删除字段的迁移。

这样做的好处是，一旦新代码有问题，可以快速回滚应用代码，而不是被数据库结构删除卡住。

# 生产上线流程

生产数据库迁移不应该只是一条 flask db upgrade。建议把它当成发布流程的一部分。

一个相对稳的上线清单如下：

在本地生成迁移脚本，并通过代码 review。
在测试环境从旧数据库结构执行 upgrade。
跑自动化测试和核心业务回归。
在预发环境使用接近生产的数据量验证耗时。
对高风险 DDL 生成 SQL 并评审执行计划。
生产执行前确认备份、监控、回滚预案。
先执行兼容性迁移，再发布应用代码。
发布后检查数据库版本、接口错误率、慢查询、锁等待。

这套流程的核心不是复杂，而是把风险显性化。数据库迁移一旦执行，影响的是共享状态，回滚成本通常比应用代码高。

# 零停机迁移思路

生产系统最怕的一类迁移，是“数据库结构已经变了，但旧代码或新代码有一边不兼容”。要降低这种风险，可以遵循 expand and contract 思路。

# 第一步：扩展结构

先做向后兼容的结构变更，例如新增可空字段、新增表、新增索引。这个阶段旧代码仍然可以正常运行。

def upgrade():
    op.add_column("user", sa.Column("email", sa.String(length=128), nullable=True))

1
2

# 第二步：发布兼容代码

应用代码开始写新字段，但仍然兼容旧数据。例如读取时允许 email 为空，写入时补充 email。

# 第三步：数据回填

用后台任务、脚本或批处理把历史数据补齐。大表回填要分批执行，不要在一次迁移脚本里做长时间循环。

示例思路：

last_id = 0
batch_size = 1000

while True:
    users = (
        User.query
        .filter(User.id > last_id, User.email.is_(None))
        .order_by(User.id.asc())
        .limit(batch_size)
        .all()
    )
    if not users:
        break

    for user in users:
        user.email = build_default_email(user)

    db.session.commit()
    last_id = users[-1].id

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

注意，这类数据回填脚本通常不建议直接塞进 Alembic 迁移里。结构迁移和数据修复要分清楚，尤其是数据量大、执行时间长、需要限速或重试时。

# 第四步：收缩结构

确认新代码稳定、历史数据补齐后，再做非空约束、删除旧字段、删除旧表等收缩动作。

def upgrade():
    op.alter_column("user", "email", nullable=False)

1
2

这个过程可能要跨多个版本发布，不要追求一次迁移完成所有事情。

# 回滚不是万能的

Flask-Migrate 提供 downgrade，但生产回滚不能简单理解为“执行 downgrade 就好了”。

有些变更天然不可逆：

删除字段后，字段里的历史数据已经没了。
字段类型转换失败时，原始格式可能丢失。
合并表、拆分表后，关系可能无法精确还原。
数据清洗脚本可能已经修改大量业务数据。

所以生产回滚要区分两类：

应用代码回滚：把服务回到上一个版本。
数据库结构回滚：执行反向迁移。

真实项目里，更多时候会优先让数据库变更保持向后兼容，从而允许应用代码快速回滚，而不是依赖数据库立即回滚。

# 团队协作中的迁移冲突

多人同时开发时，可能会出现多个分支各自生成迁移脚本的情况。合并后如果出现多个 head，可以用下面命令查看：

flask --app app.server:app db heads

如果确实有多个迁移头，需要创建 merge revision：

flask --app app.server:app db merge -m "merge migration heads" <head1> <head2>

团队里建议约定几条规则：

每个 PR 只包含和本需求相关的迁移。
迁移文件命名说明要清晰。
不手动修改已经合并到主干并部署过的迁移文件。
如果迁移已经进入共享环境，需要新增迁移修正，而不是改历史。
CI 中检查模型和迁移是否一致。

不要把迁移冲突当成普通代码冲突随手改。迁移文件描述的是数据库版本链，乱改会导致不同环境状态分叉。

# CI/CD 中怎么放迁移

迁移可以放进发布流水线，但要分级处理。

低风险迁移可以自动执行，例如：

新增可空字段。
新增小表。
新增低风险普通索引。
修改注释类元数据。

高风险迁移建议人工审批或单独执行，例如：

大表新增索引。
删除字段或删除表。
修改字段类型。
大规模数据回填。
新增非空约束。
影响热点表的 DDL。

流水线里至少可以做这些检查：

flask --app app.server:app db current
flask --app app.server:app db heads
flask --app app.server:app db upgrade
pytest

1
2
3
4

如果使用临时数据库跑 CI，可以每次从空库开始执行完整迁移链，确保迁移历史没有断裂。

# 推荐目录结构

一个 Flask 后端项目可以采用类似结构：

app/
  server.py
internal/
  extensions.py
  models/
    user.py
  migration/
    env.py
    script.py.mako
    versions/
      20260805_add_email_to_user.py
tests/

1
2
3
4
5
6
7
8
9
10
11
12

如果迁移目录不放在默认的 migrations，初始化 Migrate 时要固定传入 directory，后续命令也要使用同一个应用配置加载。

# 迁移脚本模板建议

一份生产可读的迁移脚本，最好让 review 人一眼知道它做了什么、有什么风险。

"""add email to user

Revision ID: 20260805_add_email_to_user
Revises: 20260801_create_user
Create Date: 2026-08-05
"""

from alembic import op
import sqlalchemy as sa


revision = "20260805_add_email_to_user"
down_revision = "20260801_create_user"
branch_labels = None
depends_on = None


def upgrade():
    op.add_column(
        "user",
        sa.Column("email", sa.String(length=128), nullable=True),
    )
    op.create_index("ix_user_email", "user", ["email"], unique=True)


def downgrade():
    op.drop_index("ix_user_email", table_name="user")
    op.drop_column("user", "email")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

如果某次迁移存在明显风险，可以在脚本里加非常短的注释说明前置条件，例如“历史数据已由 backfill job 补齐”。注释不需要多，但关键假设要写清楚。

# 常见问题

# 执行 migrate 后没有生成变化

常见原因：

ORM 模型没有被应用加载到。
db 实例和模型使用的不是同一个 SQLAlchemy 实例。
当前数据库结构已经和模型一致。
Alembic 无法识别字段重命名等复杂变化。

可以先确认模型模块是否在应用启动时被 import，再确认 flask --app app.server:app db migrate 使用的是正确的应用入口和配置。

# 迁移文件要不要提交

必须提交。迁移文件记录的是数据库结构版本历史。只提交模型代码，不提交迁移脚本，会导致其他环境无法稳定升级。

# 可以在迁移脚本里写数据修复吗

可以，但要谨慎。

少量、确定、幂等的数据修复可以写进迁移脚本。大批量、耗时长、需要限速、需要重试、依赖业务逻辑的数据回填，更适合独立脚本或后台任务。

# 生产可以直接执行 `flask db upgrade` 吗

可以，但不应该无脑执行。低风险迁移可以放入流水线自动执行，高风险迁移要先评审 SQL、确认备份和回滚预案。

# 小结

Flask-Migrate 的核心命令并不复杂：init 创建迁移环境，migrate 生成迁移脚本，upgrade 应用变更，downgrade 回滚版本。

真正需要掌握的是生产思维：迁移脚本要 review，DDL 有锁表风险，回滚不一定可逆，数据回填要分批，删除字段要延迟，应用代码和数据库结构要保持兼容。把这些规则落实到团队流程里，Flask-Migrate 才不只是一个命令行工具，而是一套可靠的数据库演进机制。

← Flask-SQLAlchemy 生产实践：配置、模型与会话管理 FAISS 向量数据库实践：从本地检索到 LangChain 生产封装 →

# Flask-Migrate 生产实践：从模型变更到数据库平滑迁移

# 迁移工具到底解决什么

# db.create_all() 的边界

# 安装与初始化

# 基础命令

# 一次标准的开发流程

# 迁移脚本必须人工检查

# 字段重命名

# 新增非空字段

# 创建索引

# 删除字段和删除表

# 生产上线流程

# 零停机迁移思路

# 第一步：扩展结构

# 第二步：发布兼容代码

# 第三步：数据回填

# 第四步：收缩结构

# 回滚不是万能的

# 团队协作中的迁移冲突

# CI/CD 中怎么放迁移

# 推荐目录结构

# 迁移脚本模板建议

# 常见问题

# 执行 migrate 后没有生成变化

# 迁移文件要不要提交

# 可以在迁移脚本里写数据修复吗

# 生产可以直接执行 flask db upgrade 吗

# 小结

# `db.create_all()` 的边界

# 生产可以直接执行 `flask db upgrade` 吗