表结构不能靠重建:给 FastAPI 接入 Alembic 迁移

摘要:把 SQLModel 的表结构变化交给 Alembic 管理,实际完成空库升级、版本查看和完整回滚,并通过统一数据库 URL 为切换 PostgreSQL 做准备。

数据库结构沿版本历史升级和回滚

图:迁移脚本记录每次结构变化,让数据库能够按版本向前升级或安全回滚。

修改 Python 模型很容易,已经装有真实数据的数据库却不会自动跟着变化。开发时删掉 SQLite 文件重来一次似乎没问题,到了共享环境和生产库,这种做法等同于删数据。

数据库结构需要自己的版本历史。下面接入 Alembic,把“创建 tasks 表”写成一条可审查的迁移,并真正跑一遍 upgrade → current → downgrade,而不只停留在配置文件层面。

先跑一遍升级和版本查看

alembic -c examples/ch10_migrations/alembic.ini upgrade head
alembic -c examples/ch10_migrations/alembic.ini current

预期数据库升级到 20260720_01 (head),并出现 tasksalembic_version 两张表。

先确认应用不再偷偷 create_all

cd 04-fastapi-beginner
source .venv/bin/activate
alembic --version

配置、环境脚本和版本脚本都在 ch10_migrations。本文应用不再调用 create_all();必须先完成迁移再访问数据接口。

把第一版表结构写成迁移

连接模型元数据

migrations/env.py 导入 Task 模型,并设置:

target_metadata = SQLModel.metadata

这让 Alembic 生成迁移时可以比较模型元数据与数据库结构。已提交的历史迁移仍是数据库演进的事实记录。

编写第一条迁移

def upgrade() -> None:
    op.create_table(...)
    op.create_index("ix_tasks_title", "tasks", ["title"])


def downgrade() -> None:
    op.drop_index("ix_tasks_title", table_name="tasks")
    op.drop_table("tasks")

upgrade 定义向前变化,downgrade 定义回退路径。迁移脚本进入 Git,和应用代码一起评审。

用环境变量覆盖 URL

environment_url = os.getenv("TASKS_DATABASE_URL")
if environment_url:
    config.set_main_option("sqlalchemy.url", environment_url)

SQLite URL 示例为 sqlite:///./ch10_tasks.db,PostgreSQL 使用 SQLAlchemy psycopg 形式:postgresql+psycopg://user:password@host:5432/tasks

模型元数据连接迁移环境并生成升级脚本

图:迁移环境读取模型元数据,比较结构后形成可执行、可审查的升级脚本。

真正执行升级、回滚和再次升级

升级、查看、回滚、再升级:

alembic -c examples/ch10_migrations/alembic.ini upgrade head
alembic -c examples/ch10_migrations/alembic.ini current
alembic -c examples/ch10_migrations/alembic.ini downgrade base
alembic -c examples/ch10_migrations/alembic.ini upgrade head

启动应用并创建任务:

fastapi dev examples/ch10_migrations/main.py
curl -X POST http://127.0.0.1:8000/tasks \
  -H "Content-Type: application/json" -d '{"title":"经过迁移的任务"}'

切换 PostgreSQL 时,在服务器和 Alembic 命令前设置同一个变量:

export TASKS_DATABASE_URL="postgresql+psycopg://postgres:postgres@localhost:5432/tasks"
alembic -c examples/ch10_migrations/alembic.ini upgrade head

本地没有 PostgreSQL 时仍可完成 SQLite 全流程和自动测试:

python -m pytest tests/test_ch10.py -q

测试在临时目录创建数据库,检查列名,再回滚并确认表消失。

模型描述现在,迁移描述如何到达现在

模型描述当前代码期望的结构,迁移描述数据库从旧状态到新状态的路径。直接修改模型不会自动、安全地改变生产数据;迁移脚本让变更顺序明确、可审查、可重复。

自动生成迁移只是草稿。重命名字段可能被识别为“删除旧列 + 新建列”,造成数据丢失,因此生成后必须阅读 upgrade 和 downgrade。

生产回滚也不应盲目执行 downgrade。若新版本已写入新格式数据,需要先评估兼容性、备份和恢复策略。

模型描述当前结构,迁移保存变化过程和历史

图:模型是当前结构的快照,迁移是数据库一步步到达当前状态的历史。

迁移脚本最容易踩的五个坑

  • 启动应用后提示 no such table:忘记先 upgrade head。
  • Alembic 和应用使用不同 URL:迁移了一个库,服务连接另一个库。
  • 生成空迁移:env.py 没有导入模型,元数据中没有表。
  • 直接修改已在环境执行过的历史迁移:不同环境会出现不可预测分叉。
  • 未审查自动生成的 drop 操作:可能造成不可恢复的数据损失。

给 tasks 表增加第二次结构变化

  1. 新增 created_at 模型字段,生成第二条 migration 并审查。
  2. 连续执行两次 upgrade head,验证迁移幂等状态。
  3. 把应用 URL 与 Alembic URL 故意设成不同文件,定位 no such table。
  4. 为第二条迁移增加升级和回滚测试。
  5. 写一份“上线迁移前备份和验证”清单。

最后,用测试固定这次改动

手动请求通过后,运行与本文对应的自动化测试:

python -m pytest tests/test_ch10.py -q
python tests/validate_course.py

写在最后

应用已经具备可演进的数据层。下一篇在数据库概念之上加入用户身份、密码哈希和 JWT,让任务真正属于不同调用者。