1. 什么是 Alembic为什么需要它Alembic是由 SQLAlchemy 的作者 Mike Bayer 编写的数据库迁移工具。你可以把它理解成「数据库的 Git」——它跟踪数据库结构Schema的变化让你可以在不同版本之间前进和后退。没有 Alembic 时的痛点假设你在开发一个博客项目一开始users表只有id和name后来需要加一个email字段。你可能会手动登录数据库执行ALTER TABLE users ADD COLUMN email VARCHAR(255);告诉团队成员「记得去改一下数据库」生产环境上线时再手动执行一遍这种方式的问题显而易见容易遗漏、无法追踪、无法回滚。有了 Alembic 之后所有数据库结构的变更都以迁移脚本的形式保存在代码仓库中团队成员只需执行一条命令alembic upgrade head即可同步最新结构可以随时回滚到任意历史版本支持自动检测 SQLAlchemy Model 的变化自动生成迁移脚本2. 安装与初始化安装pip install alembic sqlalchemy如果你使用 PostgreSQLpip install psycopg2-binary初始化 Alembic 环境在项目根目录下执行以下命令完成初始化alembic init alembic执行后项目目录结构如下myproject/ ├── alembic/ │ ├── env.py # 核心配置脚本每次运行迁移都会执行 │ ├── README │ ├── script.py.mako # 迁移脚本的模板文件 │ └── versions/ # 存放所有迁移版本文件的目录 │ └── 这里会生成迁移脚本 ├── alembic.ini # Alembic 的主配置文件 └── app/ └── models.py # 你的 SQLAlchemy 模型理解各个文件的作用alembic.ini— 主配置文件主要配置数据库连接 URL[alembic] # 迁移脚本存放位置 script_location alembic # 数据库连接 URL sqlalchemy.url sqlalchemy.url postgresqlpsycopg2://username:passworddb_ip:5432/db_name # 省略其他...env.py— 这是 Alembic 的核心运行脚本。每次你执行任何alembic命令这个文件都会被执行。它负责建立数据库连接配置迁移上下文区分「在线模式」直接执行 SQL和「离线模式」生成 SQL 文件versions/— 存放所有迁移文件。文件名格式类似3512b954651e_add_users_table.py使用部分 GUID 而非递增整数来命名这样可以更好地支持分支合并。3. 基本使用1 创建你的第一个 Model先创建一个简单的 SQLAlchemy 模型文件# app/models.py from sqlalchemy import Column, Integer, String, DateTime from sqlalchemy.orm import declarative_base import datetime Base declarative_base() class User(Base): __tablename__ users id Column(Integer, primary_keyTrue, indexTrue) name Column(String(50), nullableFalse) email Column(String(200), nullableFalse, uniqueTrue) created_at Column(DateTime, defaultdatetime.datetime.utcnow)2 将Model 与 Alembic关联修改alembic/env.py让 Alembic 知道你的 Model 在哪里# alembic/env.py 中找到这行 # target_metadata None # 替换为 import sys import os from app.models import Base # 导入你的 Base target_metadata Base.metadata # 告诉 Alembic 用哪个 metadata 来检测变化 # 或 target_metadata [Base.metadata, ]新手常见错误忘记把target_metadata None改成你的Base.metadata导致自动生成的迁移脚本是空的。3 创建第一个迁移脚本alembic revision --autogenerate-m create users table执行后alembic/versions/目录下会生成类似这样的文件# alembic/versions/673707e9f34b_create_users_table.py create users table Revision ID: 673707e9f34b Revises: Create Date: 2026-03-20 08:57:19.170729 from typing import Sequence, Union from alembic import op import sqlalchemy as sa # 版本标识Alembic 用这个来追踪当前在哪个版本 revision: str 673707e9f34b down_revision: Union[str, Sequence[str], None] None # 前一个版本的 revision ID None 表示这是第一个迁移 branch_labels: Union[str, Sequence[str], None] None depends_on: Union[str, Sequence[str], None] None def upgrade() - None: Upgrade schema. # ### commands auto generated by Alembic - please adjust! ### op.create_table(users, sa.Column(id, sa.Integer(), nullableFalse), sa.Column(name, sa.String(length50), nullableFalse), sa.Column(email, sa.String(length200), nullableFalse), sa.Column(created_at, sa.DateTime(), nullableTrue), sa.PrimaryKeyConstraint(id), sa.UniqueConstraint(email) ) op.create_index(op.f(ix_users_id), users, [id], uniqueFalse) # ### end Alembic commands ### def downgrade() - None: Downgrade schema. # ### commands auto generated by Alembic - please adjust! ### op.drop_index(op.f(ix_users_id), table_nameusers) op.drop_table(users) # ### end Alembic commands ###每个迁移脚本包含两个函数upgrade()— 应用变更往新版本走downgrade()— 撤销变更往旧版本退4 执行迁移升级到最新版本alembic upgrade headhead表示升级到最新版本。执行后Alembic 会连接数据库检查alembic_version表首次运行时自动创建确认当前所处的版本依次执行所有尚未应用的迁移脚本中的upgrade()函数如果你想升级到某个特定版本alembic upgrade 3512b954651e # Revision ID或者升级 N 步alembic upgrade 25 回滚迁移回滚到上一个版本alembic downgrade -1回滚到某个特定版本alembic downgrade 3512b954651e # alembic downgrade 3512b954651e回滚到最初状态撤销所有迁移alembic downgrade base4. 自动生成迁移脚本autogenerate手动编写迁移脚本很麻烦Alembic 的杀手锏功能是自动生成autogenerate。其工作原理是将你的 SQLAlchemy Model 定义目标状态与数据库中实际的表结构当前状态进行比对自动生成一个描述两者「差异」的迁移脚本。使用示例假设你在user.py中新增了一个phone字段# app/models/User.py from sqlalchemy import Column, Integer, String, DateTime from sqlalchemy.orm import declarative_base import datetime Base declarative_base() class User(Base): __tablename__ users # 命名约束定义表名建议添加 id Column(Integer, primary_keyTrue, indexTrue) name Column(String(50), nullableFalse) email Column(String(200), nullableFalse, uniqueTrue) phone Column(String(20), nullableFalse, uniqueTrue) # 新增phone字段 created_at Column(DateTime, defaultdatetime.datetime.utcnow)然后执行自动生成命令alembic revision --autogenerate -m add phone field to user tableAlembic 会自动生成# alembic/versions/7f29ab2b2004_add_phone_field_to_user_table.py add phone field to user table Revision ID: 7f29ab2b2004 Revises: 673707e9f34b Create Date: 2026-03-20 09:12:51.346000 from typing import Sequence, Union from alembic import op import sqlalchemy as sa # revision identifiers, used by Alembic. revision: str 7f29ab2b2004 down_revision: Union[str, Sequence[str], None] 673707e9f34b branch_labels: Union[str, Sequence[str], None] None depends_on: Union[str, Sequence[str], None] None def upgrade() - None: Upgrade schema. # ### commands auto generated by Alembic - please adjust! ### op.add_column(users, sa.Column(phone, sa.String(length20), nullableFalse)) op.create_unique_constraint(None, users, [phone]) # ### end Alembic commands ### def downgrade() - None: Downgrade schema. # ### commands auto generated by Alembic - please adjust! ### op.drop_constraint(None, users, type_unique) op.drop_column(users, phone) # ### end Alembic commands ###最后执行升级命令使变更生效alembic upgrade head查看当前版本和历史查看当前数据库所在的版本alembic current输出示例INFO [alembic.runtime.migration] Context impl PostgresqlImpl. INFO [alembic.runtime.migration] Will assume transactional DDL. 7f29ab2b2004 (head)查看所有迁移历史alembic history输出示例673707e9f34b - 7f29ab2b2004 (head), add phone field to user table base - 673707e9f34b, create users tableautogenerate 能检测哪些变化默认支持检测✅新增 / 删除表新增 / 删除列列的 nullable 状态变化基本索引变化显式命名的唯一约束变化基本外键约束变化无法检测 ❌变化类型原因表名重命名会被识别为「删除旧表 新建新表」需手动改成 rename 操作列名重命名同上会被识别为「删除旧列 新增新列」直接导致数据丢失匿名约束没有名字的约束无法被可靠追踪不原生支持 ENUM 的数据库上的枚举类型SQLite 等数据库用CHAR CHECK模拟枚举Alembic 无法从反射结果中判断这是否是一个枚举存储过程、函数、触发器等超出 Alembic 的管理范围自动生成的迁移脚本永远不是 100% 完美的每次生成后都必须人工审核这是官方文档明确强调的原则autogenerate 的目标是「辅助」而非「替代」人工判断。5.结合Fastapi SQLModel使用SQLModel是为FastApi设计的与数据库交互的库基于 Python 类型注解并由 Pydantic 和 SQLAlchemy 驱动。1. 安装依赖pip install fastapi[standard] sqlmodel2. 定义模型# app/models/User.py from sqlmodel import SQLModel, Field from datetime import datetime, timezone class UserBase(SQLModel): 用户基础模型 公共字段基类被 Create / Response 等 Schema 复用 name: str Field(indexTrue, description用户名) email: str Field(indexTrue, uniqueTrue, description用户邮箱) created_at: datetime Field(lambda: datetime.now(timezone.utc), description创建时间) class User(UserBase, tableTrue): 用户模型 映射数据库表 users __tablename__ users id: int Field(defaultNone, primary_keyTrue) # request schema class UserCreate(UserBase): 创建用户的请求体 pass # response schema class UserRead(UserBase): 返回给前端的响应体 id: int3. 修改迁移脚本模板文件script.py.mako是所有迁移脚本的模板文件。由于 SQLModel 生成的迁移脚本默认不会导入sqlmodel模块但生成的脚本内容通常又依赖它因此推荐在模板中统一添加这个导入避免每次生成后都要手动修改。# alembic/script.py.mako # 添加 **import sqlmodel** **${message} Revision ID: ${up_revision} Revises: ${down_revision | comma,n} Create Date: ${create_date} from typing import Sequence, Union from alembic import op import sqlalchemy as sa import sqlmodel # 新增代码 ${imports if imports else } # revision identifiers, used by Alembic. revision: str ${repr(up_revision)} down_revision: Union[str, Sequence[str], None] ${repr(down_revision)} branch_labels: Union[str, Sequence[str], None] ${repr(branch_labels)} depends_on: Union[str, Sequence[str], None] ${repr(depends_on)} def upgrade() - None: Upgrade schema. ${upgrades if upgrades else pass} def downgrade() - None: Downgrade schema. ${downgrades if downgrades else pass}**4. 配置 env.pyfrom app.models.User import User target_metadata [ User.metadata, ] # 省略其他5.生成迁移脚本alembic revision --autogenerate -m create users table生成的迁移脚本如下create users table Revision ID: 54b2d338fcc9 Revises: Create Date: 2026-03-20 10:11:18.524007 from typing import Sequence, Union from alembic import op import sqlalchemy as sa import sqlmodel # 如果不修改mako文件这里需要手动引入sqlmodel # revision identifiers, used by Alembic. revision: str 54b2d338fcc9 down_revision: Union[str, Sequence[str], None] None branch_labels: Union[str, Sequence[str], None] None depends_on: Union[str, Sequence[str], None] None def upgrade() - None: Upgrade schema. # ### commands auto generated by Alembic - please adjust! ### op.create_table(users, sa.Column(name, sqlmodel.sql.sqltypes.AutoString(), nullableFalse), sa.Column(email, sqlmodel.sql.sqltypes.AutoString(), nullableFalse), sa.Column(created_at, sa.DateTime(), nullableFalse), sa.Column(id, sa.Integer(), nullableFalse), sa.PrimaryKeyConstraint(id) ) op.create_index(op.f(ix_users_email), users, [email], uniqueTrue) op.create_index(op.f(ix_users_name), users, [name], uniqueFalse) # ### end Alembic commands ### def downgrade() - None: Downgrade schema. # ### commands auto generated by Alembic - please adjust! ### op.drop_index(op.f(ix_users_name), table_nameusers) op.drop_index(op.f(ix_users_email), table_nameusers) op.drop_table(users) # ### end Alembic commands ###6. 多环境管理实际项目中通常需要分别管理本地开发、预发布staging和生产production三套环境的数据库每套环境都有各自独立的连接配置和迁移历史。Alembic 支持通过-n参数指定目标环境来实现这一需求。1. 修改alembic.ini[alembic] script_location alembic/local sqlalchemy.url postgresqlpsycopg2://username:paasowrdlocal_db_ip:5432/local_db_name [staging] script_location alembic/staging sqlalchemy.url postgresqlpsycopg2://username:paasowrdstaging_db_ip:5432/staging_db_name [production] script_location alembic/production sqlalchemy.url postgresqlpsycopg2://username:paasowrdproduction_db_ip:5432/production_db_ip_name [DEFAULT] prepend_sys_path . version_path_separator os # 省略其他配置2. 初始化alembic环境alembic init alembic/local alembic -n staging init alembic/staging alembic -n production init alembic/production生成的目录结构如下your_project/ ├── alembic/ │ ├── local/ │ │ └── versions/ │ ├── production/ │ │ └── versions/ │ └── staging/ │ └── versions/ ├── app/ │ └── models/ └── alembic.ini3. 针对不同环境执行迁移# local默认环境无需 -n 参数 alembic revision --autogenerate -m add users table alembic upgrade head # staging alembic -n staging revision --autogenerate -m add users table alembic -n staging upgrade head # production alembic -n production revision --autogenerate -m add users table alembic -n production upgrade head7. 生产环境最佳实践1. 迁移前务必备份数据库在对生产数据库执行任何迁移之前应当先完整备份以防迁移失败时可以快速恢复。# 全量备份 pg_dump -U postgres -F c -v -f test_db_$(date %Y%m%d_%H%M%S).backup test_db # 若需要恢复 psql -U postgres -c CREATE DATABASE test_db_restore; pg_restore -U postgres -d test_db_restore -v --clean test_db_20260320_031055.backup2. 生产环境的数据库 URL 不应硬编码在配置文件中将数据库连接地址明文写入alembic.ini存在安全隐患推荐改用pydantic_settings从环境变量中读取。第一步删除alembic.ini中的sqlalchemy.url# 删除sqlalchemy.url [production] script_location alembic/production第二步新增 pydantic 配置文件# app/core/settings.py from pydantic_settings import BaseSettings class Settings(BaseSettings): DATABASE_URL: str settings Settings()第三步修改 production 的env.py# alembic/production/env.py from app.core.settings import settings # 使用环境变量里的databse url配置 config.set_main_option(sqlalchemy.url, settings.DATABASE_URL)第四步设置环境变量后再执行迁移命令# 设置迁URL后再执行迁移命令 export DATABASE_URL你的DATABSE_URL alembic -n production revision --autogenerate -m create users table8. 枚举类型的正确处理方式枚举Enum字段的迁移处理是一个容易踩坑的地方值得单独说明。为什么不推荐 PostgreSQL 原生 EnumAlembic 虽然支持 PostgreSQL 的原生 Enum 类型但实际使用中存在以下问题PostgreSQL 原生枚举类型只能追加值不能删除如果需要修改枚举值只能重建整个枚举类型操作繁琐Alembic 无法自动识别枚举值的变化需要完全手动处理迁移脚本# app/models/User.py from enum import Enum from sqlmodel import SQLModel, Field from datetime import datetime, timezone class UserStatus(str, Enum): 用户状态枚举 ACTIVE active BANNED banned DELETED deleted class UserBase(SQLModel): # 省略其他 status: UserStatus Field(defaultUserStatus.ACTIVE, description用户状态) # 使用原生Enum❌不推荐推荐方案varchar CHECK 约束使用sa_column将字段定义为varchar类型并通过 CHECK 约束限制合法值范围。这样 Alembic 就能检测到枚举值的变化并自动生成迁移脚本。# app/models/User.py from sqlalchemy import Column, Enum as SaEnum class UserBase(SQLModel): status: UserStatus | None Field( defaultNone, description用户状态, sa_columnColumn(SaEnum( UserStatus, nameuser_status, create_constraintTrue, native_enumFalse, nullableTrue )) ) # 省略其他生成的迁移脚本如下# alembic/production/versions/568f7d9f2553_change_status_field.py def upgrade() - None: Upgrade schema. # ### commands auto generated by Alembic - please adjust! ### op.add_column( users, sa.Column( status, sa.Enum(ACTIVE, BANNED, DELETED, nameuser_status, native_enumFalse, create_constraintTrue ), nullableTrue ) ) # ### end Alembic commands ### def downgrade() - None: Downgrade schema. # ### commands auto generated by Alembic - please adjust! ### op.drop_column(users, status) # ### end Alembic commands ###查看users表的约束可以看到status字段类型为varchar并关联了名为user_status的 CHECK 约束。这种方案最大的优势是当你修改UserStatus枚举时Alembic 能够自动识别变化并生成对应的迁移脚本。场景一新增枚举值# app/models/User.py class UserStatus(str, Enum): 用户状态枚举 ACTIVE active BANNED banned DELETED deleted IN_ACTIVE in_active # 新增自动生成的迁移脚本如下# alembic/production/versions/568f7d9f2553_change_status_field.py def upgrade() - None: Upgrade schema. # ### commands auto generated by Alembic - please adjust! ### op.alter_column(users, status, existing_typesa.VARCHAR(length7), type_sa.Enum(ACTIVE, BANNED, DELETED, IN_ACTIVE, nameuser_status, native_enumFalse, create_constraintTrue), existing_nullableTrue) # ### end Alembic commands ### def downgrade() - None: Downgrade schema. # ### commands auto generated by Alembic - please adjust! ### op.alter_column(users, status, existing_typesa.Enum(ACTIVE, BANNED, DELETED, IN_ACTIVE, nameuser_status, native_enumFalse, create_constraintTrue), type_sa.VARCHAR(length7), existing_nullableTrue) # ### end Alembic commands ###注意直接执行上述脚本会报错sqlalchemy.exc.ProgrammingError: (psycopg2.errors.DuplicateObject) constraint user_status for relation users already exists原因是 Alembic 在alter_column时会尝试重新创建 CHECK 约束但旧的约束还未删除。需要手动在脚本中先删除旧约束# alembic/production/versions/568f7d9f2553_change_status_field.py def upgrade() - None: Upgrade schema. # ### commands auto generated by Alembic - please adjust! ### op.drop_constraint(user_status, users, type_check) # 手动添加先删除旧约束 op.alter_column(users, status, existing_typesa.VARCHAR(length7), type_sa.Enum(ACTIVE, BANNED, DELETED, IN_ACTIVE, nameuser_status, native_enumFalse, create_constraintTrue), existing_nullableTrue) # ### end Alembic commands ### def downgrade() - None: Downgrade schema. # ### commands auto generated by Alembic - please adjust! ### op.drop_constraint(user_status, users, type_check) # # 手动添加先删除旧约束 op.alter_column(users, status, existing_typesa.Enum(ACTIVE, BANNED, DELETED, IN_ACTIVE, nameuser_status, native_enumFalse, create_constraintTrue), type_sa.VARCHAR(length7), existing_nullableTrue) # ### end Alembic commands ###场景二修改已有枚举值 Alembic 只能检测枚举结构的增删变化新增或删除某个枚举值对于修改枚举值本身如将BANNED改为DISABLED则无法感知需要完全手动编写迁移脚本。# app/models/User.py class UserStatus(str, Enum): 用户状态枚举 ACTIVE active DISABLED disabled # 将BANNED修改为DISABLED DELETED deleted IN_ACTIVE in_active对于这类情况需要手动编写迁移脚本包括先删除旧约束、更新现有数据中的历史枚举值、再创建新约束# alembic/productionversions/0d46a300af09_change_status_banned_to_disabled.py def upgrade() - None: Upgrade schema. # ### commands auto generated by Alembic - please adjust! ### op.drop_constraint(user_status, users, type_check) op.execute(UPDATE users SET status IN_ACTIVE WHERE status IS_ACTIVE) op.execute(UPDATE users SET status DISABLED WHERE status BANNED) op.create_check_constraint( user_status, users, status IN (ACTIVE, DISABLED, DELETED, IN_ACTIVE) ) # ### end Alembic commands ### def downgrade() - None: Downgrade schema. # ### commands auto generated by Alembic - please adjust! ### op.drop_constraint(user_status, users, type_check) op.execute(UPDATE users SET status BANNED WHERE status DISABLED) op.create_check_constraint( user_status, users, status IN (ACTIVE, BANNED, DELETED, IN_ACTIVE) ) # ### end Alembic commands ### Alembic入门教程
Alembic入门教程
发布时间：2026/7/15 1:23:47
1. 什么是 Alembic为什么需要它Alembic是由 SQLAlchemy 的作者 Mike Bayer 编写的数据库迁移工具。你可以把它理解成「数据库的 Git」——它跟踪数据库结构Schema的变化让你可以在不同版本之间前进和后退。没有 Alembic 时的痛点假设你在开发一个博客项目一开始users表只有id和name后来需要加一个email字段。你可能会手动登录数据库执行ALTER TABLE users ADD COLUMN email VARCHAR(255);告诉团队成员「记得去改一下数据库」生产环境上线时再手动执行一遍这种方式的问题显而易见容易遗漏、无法追踪、无法回滚。有了 Alembic 之后所有数据库结构的变更都以迁移脚本的形式保存在代码仓库中团队成员只需执行一条命令alembic upgrade head即可同步最新结构可以随时回滚到任意历史版本支持自动检测 SQLAlchemy Model 的变化自动生成迁移脚本2. 安装与初始化安装pip install alembic sqlalchemy如果你使用 PostgreSQLpip install psycopg2-binary初始化 Alembic 环境在项目根目录下执行以下命令完成初始化alembic init alembic执行后项目目录结构如下myproject/ ├── alembic/ │ ├── env.py # 核心配置脚本每次运行迁移都会执行 │ ├── README │ ├── script.py.mako # 迁移脚本的模板文件 │ └── versions/ # 存放所有迁移版本文件的目录 │ └── 这里会生成迁移脚本 ├── alembic.ini # Alembic 的主配置文件 └── app/ └── models.py # 你的 SQLAlchemy 模型理解各个文件的作用alembic.ini— 主配置文件主要配置数据库连接 URL[alembic] # 迁移脚本存放位置 script_location alembic # 数据库连接 URL sqlalchemy.url sqlalchemy.url postgresqlpsycopg2://username:passworddb_ip:5432/db_name # 省略其他...env.py— 这是 Alembic 的核心运行脚本。每次你执行任何alembic命令这个文件都会被执行。它负责建立数据库连接配置迁移上下文区分「在线模式」直接执行 SQL和「离线模式」生成 SQL 文件versions/— 存放所有迁移文件。文件名格式类似3512b954651e_add_users_table.py使用部分 GUID 而非递增整数来命名这样可以更好地支持分支合并。3. 基本使用1 创建你的第一个 Model先创建一个简单的 SQLAlchemy 模型文件# app/models.py from sqlalchemy import Column, Integer, String, DateTime from sqlalchemy.orm import declarative_base import datetime Base declarative_base() class User(Base): __tablename__ users id Column(Integer, primary_keyTrue, indexTrue) name Column(String(50), nullableFalse) email Column(String(200), nullableFalse, uniqueTrue) created_at Column(DateTime, defaultdatetime.datetime.utcnow)2 将Model 与 Alembic关联修改alembic/env.py让 Alembic 知道你的 Model 在哪里# alembic/env.py 中找到这行 # target_metadata None # 替换为 import sys import os from app.models import Base # 导入你的 Base target_metadata Base.metadata # 告诉 Alembic 用哪个 metadata 来检测变化 # 或 target_metadata [Base.metadata, ]新手常见错误忘记把target_metadata None改成你的Base.metadata导致自动生成的迁移脚本是空的。3 创建第一个迁移脚本alembic revision --autogenerate-m create users table执行后alembic/versions/目录下会生成类似这样的文件# alembic/versions/673707e9f34b_create_users_table.py create users table Revision ID: 673707e9f34b Revises: Create Date: 2026-03-20 08:57:19.170729 from typing import Sequence, Union from alembic import op import sqlalchemy as sa # 版本标识Alembic 用这个来追踪当前在哪个版本 revision: str 673707e9f34b down_revision: Union[str, Sequence[str], None] None # 前一个版本的 revision ID None 表示这是第一个迁移 branch_labels: Union[str, Sequence[str], None] None depends_on: Union[str, Sequence[str], None] None def upgrade() - None: Upgrade schema. # ### commands auto generated by Alembic - please adjust! ### op.create_table(users, sa.Column(id, sa.Integer(), nullableFalse), sa.Column(name, sa.String(length50), nullableFalse), sa.Column(email, sa.String(length200), nullableFalse), sa.Column(created_at, sa.DateTime(), nullableTrue), sa.PrimaryKeyConstraint(id), sa.UniqueConstraint(email) ) op.create_index(op.f(ix_users_id), users, [id], uniqueFalse) # ### end Alembic commands ### def downgrade() - None: Downgrade schema. # ### commands auto generated by Alembic - please adjust! ### op.drop_index(op.f(ix_users_id), table_nameusers) op.drop_table(users) # ### end Alembic commands ###每个迁移脚本包含两个函数upgrade()— 应用变更往新版本走downgrade()— 撤销变更往旧版本退4 执行迁移升级到最新版本alembic upgrade headhead表示升级到最新版本。执行后Alembic 会连接数据库检查alembic_version表首次运行时自动创建确认当前所处的版本依次执行所有尚未应用的迁移脚本中的upgrade()函数如果你想升级到某个特定版本alembic upgrade 3512b954651e # Revision ID或者升级 N 步alembic upgrade 25 回滚迁移回滚到上一个版本alembic downgrade -1回滚到某个特定版本alembic downgrade 3512b954651e # alembic downgrade 3512b954651e回滚到最初状态撤销所有迁移alembic downgrade base4. 自动生成迁移脚本autogenerate手动编写迁移脚本很麻烦Alembic 的杀手锏功能是自动生成autogenerate。其工作原理是将你的 SQLAlchemy Model 定义目标状态与数据库中实际的表结构当前状态进行比对自动生成一个描述两者「差异」的迁移脚本。使用示例假设你在user.py中新增了一个phone字段# app/models/User.py from sqlalchemy import Column, Integer, String, DateTime from sqlalchemy.orm import declarative_base import datetime Base declarative_base() class User(Base): __tablename__ users # 命名约束定义表名建议添加 id Column(Integer, primary_keyTrue, indexTrue) name Column(String(50), nullableFalse) email Column(String(200), nullableFalse, uniqueTrue) phone Column(String(20), nullableFalse, uniqueTrue) # 新增phone字段 created_at Column(DateTime, defaultdatetime.datetime.utcnow)然后执行自动生成命令alembic revision --autogenerate -m add phone field to user tableAlembic 会自动生成# alembic/versions/7f29ab2b2004_add_phone_field_to_user_table.py add phone field to user table Revision ID: 7f29ab2b2004 Revises: 673707e9f34b Create Date: 2026-03-20 09:12:51.346000 from typing import Sequence, Union from alembic import op import sqlalchemy as sa # revision identifiers, used by Alembic. revision: str 7f29ab2b2004 down_revision: Union[str, Sequence[str], None] 673707e9f34b branch_labels: Union[str, Sequence[str], None] None depends_on: Union[str, Sequence[str], None] None def upgrade() - None: Upgrade schema. # ### commands auto generated by Alembic - please adjust! ### op.add_column(users, sa.Column(phone, sa.String(length20), nullableFalse)) op.create_unique_constraint(None, users, [phone]) # ### end Alembic commands ### def downgrade() - None: Downgrade schema. # ### commands auto generated by Alembic - please adjust! ### op.drop_constraint(None, users, type_unique) op.drop_column(users, phone) # ### end Alembic commands ###最后执行升级命令使变更生效alembic upgrade head查看当前版本和历史查看当前数据库所在的版本alembic current输出示例INFO [alembic.runtime.migration] Context impl PostgresqlImpl. INFO [alembic.runtime.migration] Will assume transactional DDL. 7f29ab2b2004 (head)查看所有迁移历史alembic history输出示例673707e9f34b - 7f29ab2b2004 (head), add phone field to user table base - 673707e9f34b, create users tableautogenerate 能检测哪些变化默认支持检测✅新增 / 删除表新增 / 删除列列的 nullable 状态变化基本索引变化显式命名的唯一约束变化基本外键约束变化无法检测 ❌变化类型原因表名重命名会被识别为「删除旧表 新建新表」需手动改成 rename 操作列名重命名同上会被识别为「删除旧列 新增新列」直接导致数据丢失匿名约束没有名字的约束无法被可靠追踪不原生支持 ENUM 的数据库上的枚举类型SQLite 等数据库用CHAR CHECK模拟枚举Alembic 无法从反射结果中判断这是否是一个枚举存储过程、函数、触发器等超出 Alembic 的管理范围自动生成的迁移脚本永远不是 100% 完美的每次生成后都必须人工审核这是官方文档明确强调的原则autogenerate 的目标是「辅助」而非「替代」人工判断。5.结合Fastapi SQLModel使用SQLModel是为FastApi设计的与数据库交互的库基于 Python 类型注解并由 Pydantic 和 SQLAlchemy 驱动。1. 安装依赖pip install fastapi[standard] sqlmodel2. 定义模型# app/models/User.py from sqlmodel import SQLModel, Field from datetime import datetime, timezone class UserBase(SQLModel): 用户基础模型 公共字段基类被 Create / Response 等 Schema 复用 name: str Field(indexTrue, description用户名) email: str Field(indexTrue, uniqueTrue, description用户邮箱) created_at: datetime Field(lambda: datetime.now(timezone.utc), description创建时间) class User(UserBase, tableTrue): 用户模型 映射数据库表 users __tablename__ users id: int Field(defaultNone, primary_keyTrue) # request schema class UserCreate(UserBase): 创建用户的请求体 pass # response schema class UserRead(UserBase): 返回给前端的响应体 id: int3. 修改迁移脚本模板文件script.py.mako是所有迁移脚本的模板文件。由于 SQLModel 生成的迁移脚本默认不会导入sqlmodel模块但生成的脚本内容通常又依赖它因此推荐在模板中统一添加这个导入避免每次生成后都要手动修改。# alembic/script.py.mako # 添加 **import sqlmodel** **${message} Revision ID: ${up_revision} Revises: ${down_revision | comma,n} Create Date: ${create_date} from typing import Sequence, Union from alembic import op import sqlalchemy as sa import sqlmodel # 新增代码 ${imports if imports else } # revision identifiers, used by Alembic. revision: str ${repr(up_revision)} down_revision: Union[str, Sequence[str], None] ${repr(down_revision)} branch_labels: Union[str, Sequence[str], None] ${repr(branch_labels)} depends_on: Union[str, Sequence[str], None] ${repr(depends_on)} def upgrade() - None: Upgrade schema. ${upgrades if upgrades else pass} def downgrade() - None: Downgrade schema. ${downgrades if downgrades else pass}**4. 配置 env.pyfrom app.models.User import User target_metadata [ User.metadata, ] # 省略其他5.生成迁移脚本alembic revision --autogenerate -m create users table生成的迁移脚本如下create users table Revision ID: 54b2d338fcc9 Revises: Create Date: 2026-03-20 10:11:18.524007 from typing import Sequence, Union from alembic import op import sqlalchemy as sa import sqlmodel # 如果不修改mako文件这里需要手动引入sqlmodel # revision identifiers, used by Alembic. revision: str 54b2d338fcc9 down_revision: Union[str, Sequence[str], None] None branch_labels: Union[str, Sequence[str], None] None depends_on: Union[str, Sequence[str], None] None def upgrade() - None: Upgrade schema. # ### commands auto generated by Alembic - please adjust! ### op.create_table(users, sa.Column(name, sqlmodel.sql.sqltypes.AutoString(), nullableFalse), sa.Column(email, sqlmodel.sql.sqltypes.AutoString(), nullableFalse), sa.Column(created_at, sa.DateTime(), nullableFalse), sa.Column(id, sa.Integer(), nullableFalse), sa.PrimaryKeyConstraint(id) ) op.create_index(op.f(ix_users_email), users, [email], uniqueTrue) op.create_index(op.f(ix_users_name), users, [name], uniqueFalse) # ### end Alembic commands ### def downgrade() - None: Downgrade schema. # ### commands auto generated by Alembic - please adjust! ### op.drop_index(op.f(ix_users_name), table_nameusers) op.drop_index(op.f(ix_users_email), table_nameusers) op.drop_table(users) # ### end Alembic commands ###6. 多环境管理实际项目中通常需要分别管理本地开发、预发布staging和生产production三套环境的数据库每套环境都有各自独立的连接配置和迁移历史。Alembic 支持通过-n参数指定目标环境来实现这一需求。1. 修改alembic.ini[alembic] script_location alembic/local sqlalchemy.url postgresqlpsycopg2://username:paasowrdlocal_db_ip:5432/local_db_name [staging] script_location alembic/staging sqlalchemy.url postgresqlpsycopg2://username:paasowrdstaging_db_ip:5432/staging_db_name [production] script_location alembic/production sqlalchemy.url postgresqlpsycopg2://username:paasowrdproduction_db_ip:5432/production_db_ip_name [DEFAULT] prepend_sys_path . version_path_separator os # 省略其他配置2. 初始化alembic环境alembic init alembic/local alembic -n staging init alembic/staging alembic -n production init alembic/production生成的目录结构如下your_project/ ├── alembic/ │ ├── local/ │ │ └── versions/ │ ├── production/ │ │ └── versions/ │ └── staging/ │ └── versions/ ├── app/ │ └── models/ └── alembic.ini3. 针对不同环境执行迁移# local默认环境无需 -n 参数 alembic revision --autogenerate -m add users table alembic upgrade head # staging alembic -n staging revision --autogenerate -m add users table alembic -n staging upgrade head # production alembic -n production revision --autogenerate -m add users table alembic -n production upgrade head7. 生产环境最佳实践1. 迁移前务必备份数据库在对生产数据库执行任何迁移之前应当先完整备份以防迁移失败时可以快速恢复。# 全量备份 pg_dump -U postgres -F c -v -f test_db_$(date %Y%m%d_%H%M%S).backup test_db # 若需要恢复 psql -U postgres -c CREATE DATABASE test_db_restore; pg_restore -U postgres -d test_db_restore -v --clean test_db_20260320_031055.backup2. 生产环境的数据库 URL 不应硬编码在配置文件中将数据库连接地址明文写入alembic.ini存在安全隐患推荐改用pydantic_settings从环境变量中读取。第一步删除alembic.ini中的sqlalchemy.url# 删除sqlalchemy.url [production] script_location alembic/production第二步新增 pydantic 配置文件# app/core/settings.py from pydantic_settings import BaseSettings class Settings(BaseSettings): DATABASE_URL: str settings Settings()第三步修改 production 的env.py# alembic/production/env.py from app.core.settings import settings # 使用环境变量里的databse url配置 config.set_main_option(sqlalchemy.url, settings.DATABASE_URL)第四步设置环境变量后再执行迁移命令# 设置迁URL后再执行迁移命令 export DATABASE_URL你的DATABSE_URL alembic -n production revision --autogenerate -m create users table8. 枚举类型的正确处理方式枚举Enum字段的迁移处理是一个容易踩坑的地方值得单独说明。为什么不推荐 PostgreSQL 原生 EnumAlembic 虽然支持 PostgreSQL 的原生 Enum 类型但实际使用中存在以下问题PostgreSQL 原生枚举类型只能追加值不能删除如果需要修改枚举值只能重建整个枚举类型操作繁琐Alembic 无法自动识别枚举值的变化需要完全手动处理迁移脚本# app/models/User.py from enum import Enum from sqlmodel import SQLModel, Field from datetime import datetime, timezone class UserStatus(str, Enum): 用户状态枚举 ACTIVE active BANNED banned DELETED deleted class UserBase(SQLModel): # 省略其他 status: UserStatus Field(defaultUserStatus.ACTIVE, description用户状态) # 使用原生Enum❌不推荐推荐方案varchar CHECK 约束使用sa_column将字段定义为varchar类型并通过 CHECK 约束限制合法值范围。这样 Alembic 就能检测到枚举值的变化并自动生成迁移脚本。# app/models/User.py from sqlalchemy import Column, Enum as SaEnum class UserBase(SQLModel): status: UserStatus | None Field( defaultNone, description用户状态, sa_columnColumn(SaEnum( UserStatus, nameuser_status, create_constraintTrue, native_enumFalse, nullableTrue )) ) # 省略其他生成的迁移脚本如下# alembic/production/versions/568f7d9f2553_change_status_field.py def upgrade() - None: Upgrade schema. # ### commands auto generated by Alembic - please adjust! ### op.add_column( users, sa.Column( status, sa.Enum(ACTIVE, BANNED, DELETED, nameuser_status, native_enumFalse, create_constraintTrue ), nullableTrue ) ) # ### end Alembic commands ### def downgrade() - None: Downgrade schema. # ### commands auto generated by Alembic - please adjust! ### op.drop_column(users, status) # ### end Alembic commands ###查看users表的约束可以看到status字段类型为varchar并关联了名为user_status的 CHECK 约束。这种方案最大的优势是当你修改UserStatus枚举时Alembic 能够自动识别变化并生成对应的迁移脚本。场景一新增枚举值# app/models/User.py class UserStatus(str, Enum): 用户状态枚举 ACTIVE active BANNED banned DELETED deleted IN_ACTIVE in_active # 新增自动生成的迁移脚本如下# alembic/production/versions/568f7d9f2553_change_status_field.py def upgrade() - None: Upgrade schema. # ### commands auto generated by Alembic - please adjust! ### op.alter_column(users, status, existing_typesa.VARCHAR(length7), type_sa.Enum(ACTIVE, BANNED, DELETED, IN_ACTIVE, nameuser_status, native_enumFalse, create_constraintTrue), existing_nullableTrue) # ### end Alembic commands ### def downgrade() - None: Downgrade schema. # ### commands auto generated by Alembic - please adjust! ### op.alter_column(users, status, existing_typesa.Enum(ACTIVE, BANNED, DELETED, IN_ACTIVE, nameuser_status, native_enumFalse, create_constraintTrue), type_sa.VARCHAR(length7), existing_nullableTrue) # ### end Alembic commands ###注意直接执行上述脚本会报错sqlalchemy.exc.ProgrammingError: (psycopg2.errors.DuplicateObject) constraint user_status for relation users already exists原因是 Alembic 在alter_column时会尝试重新创建 CHECK 约束但旧的约束还未删除。需要手动在脚本中先删除旧约束# alembic/production/versions/568f7d9f2553_change_status_field.py def upgrade() - None: Upgrade schema. # ### commands auto generated by Alembic - please adjust! ### op.drop_constraint(user_status, users, type_check) # 手动添加先删除旧约束 op.alter_column(users, status, existing_typesa.VARCHAR(length7), type_sa.Enum(ACTIVE, BANNED, DELETED, IN_ACTIVE, nameuser_status, native_enumFalse, create_constraintTrue), existing_nullableTrue) # ### end Alembic commands ### def downgrade() - None: Downgrade schema. # ### commands auto generated by Alembic - please adjust! ### op.drop_constraint(user_status, users, type_check) # # 手动添加先删除旧约束 op.alter_column(users, status, existing_typesa.Enum(ACTIVE, BANNED, DELETED, IN_ACTIVE, nameuser_status, native_enumFalse, create_constraintTrue), type_sa.VARCHAR(length7), existing_nullableTrue) # ### end Alembic commands ###场景二修改已有枚举值 Alembic 只能检测枚举结构的增删变化新增或删除某个枚举值对于修改枚举值本身如将BANNED改为DISABLED则无法感知需要完全手动编写迁移脚本。# app/models/User.py class UserStatus(str, Enum): 用户状态枚举 ACTIVE active DISABLED disabled # 将BANNED修改为DISABLED DELETED deleted IN_ACTIVE in_active对于这类情况需要手动编写迁移脚本包括先删除旧约束、更新现有数据中的历史枚举值、再创建新约束# alembic/productionversions/0d46a300af09_change_status_banned_to_disabled.py def upgrade() - None: Upgrade schema. # ### commands auto generated by Alembic - please adjust! ### op.drop_constraint(user_status, users, type_check) op.execute(UPDATE users SET status IN_ACTIVE WHERE status IS_ACTIVE) op.execute(UPDATE users SET status DISABLED WHERE status BANNED) op.create_check_constraint( user_status, users, status IN (ACTIVE, DISABLED, DELETED, IN_ACTIVE) ) # ### end Alembic commands ### def downgrade() - None: Downgrade schema. # ### commands auto generated by Alembic - please adjust! ### op.drop_constraint(user_status, users, type_check) op.execute(UPDATE users SET status BANNED WHERE status DISABLED) op.create_check_constraint( user_status, users, status IN (ACTIVE, BANNED, DELETED, IN_ACTIVE) ) # ### end Alembic commands ### Alembic入门教程
)
