在 Python 后端开发中，只要项目使用了 SQLAlchemy，基本都会遇到一个问题：

数据库表结构发生变化之后，应该怎么安全地同步到数据库？

比如一开始我们有一个 users 表：

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    username = Column(String(50))

后来业务发生变化，需要增加：

1	email = Column(String(100))

如果项目还处于刚开始开发的阶段，我们当然可以直接删掉数据库重新创建。

但如果数据库中已经存在大量用户数据，这种方式显然不可行。

这时候就需要使用 Alembic。

Alembic 是 SQLAlchemy 官方推荐的数据库迁移工具，可以帮助我们管理数据库结构的变化，例如：

创建、删除数据库表
新增、删除字段
修改字段类型
创建索引和外键
升级数据库版本
回滚数据库版本
记录数据库结构的历史变化

你可以简单地把 Alembic 理解为：

数据库结构的 Git。

Git 管理代码版本，而 Alembic 管理数据库结构版本。

一、先理解数据库迁移

假设项目刚开始的时候，users 表只有两个字段：

1
2
3

users
├── id
└── username

后来增加邮箱：

users
├── id
├── username
└── email

继续开发之后可能又变成：

V1
users(id, username)

↓

V2
users(id, username, email)

↓

V3
users(id, username, email, password)

↓

V4
users(id, username, email, password, created_at)

如果只有一个本地数据库，手动执行 SQL 也不是不能用：

1	ALTER TABLE users ADD COLUMN email VARCHAR(100);

但实际项目通常至少有：

1
2
3

开发环境
测试环境
生产环境

这时候就会出现很多问题：

开发数据库执行过哪些 SQL？

测试数据库执行到哪个版本了？

生产数据库还缺哪些字段？

数据库升级失败了怎么恢复？

团队成员数据库状态是否一致？

如果这些都依靠人工维护 SQL，项目稍微复杂一点就容易混乱。

Alembic 就是为了解决这个问题。

它会为数据库的每一次结构修改生成一个迁移文件：

alembic/
└── versions/
    ├── 001_create_users.py
    ├── 002_add_email.py
    └── 003_add_created_at.py

数据库版本就可以按照顺序变化：

每个迁移文件中最重要的是两个函数：

def upgrade():
    pass


def downgrade():
    pass

upgrade() 表示数据库向新版本升级。

例如：

def upgrade():
    op.add_column(
        "users",
        sa.Column("email", sa.String(100)),
    )

而 downgrade() 表示回滚：

def downgrade():
    op.drop_column(
        "users",
        "email",
    )

所以 Alembic 的基本思路就是两个方向：

upgrade
数据库向前升级

downgrade
数据库向后回滚

二、Alembic 的安装和初始化

这里使用现在比较常见的：

FastAPI
+
SQLAlchemy 2.x
+
Alembic

作为示例。

假设项目目录：

backend/
├── app/
│   ├── models/
│   │   └── user.py
│   ├── database.py
│   └── main.py
│
├── pyproject.toml
└── README.md

如果使用 uv：

1	uv add sqlalchemy alembic

如果使用传统 pip：

1	pip install sqlalchemy alembic

检查 Alembic 是否安装成功：

1	alembic --version

然后在项目根目录执行：

1	alembic init alembic

执行完成后，目录会变成：

backend/
├── alembic/
│   ├── versions/
│   ├── env.py
│   ├── README
│   └── script.py.mako
│
├── app/
│   ├── models/
│   ├── database.py
│   └── main.py
│
├── alembic.ini
└── pyproject.toml

其中最重要的是三个地方。

1	alembic.ini

Alembic 的主配置文件。

1	alembic/env.py

Alembic 执行数据库迁移时加载的配置代码。

1	alembic/versions/

所有数据库迁移文件都会放在这里。

三、配置 SQLAlchemy 和 Alembic

先创建 SQLAlchemy 数据库配置。

app/database.py：

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker


DATABASE_URL = "sqlite:///./app.db"


engine = create_engine(
    DATABASE_URL,
    echo=True,
)


SessionLocal = sessionmaker(
    bind=engine,
    autoflush=False,
    autocommit=False,
)


class Base(DeclarativeBase):
    pass

如果是新项目，更推荐使用 SQLAlchemy 2.x 的：

1 2	class Base(DeclarativeBase): pass

而不是旧版本常见的：

1	Base = declarative_base()

然后创建一个 User 模型：

from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column

from app.database import Base


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(
        primary_key=True,
        autoincrement=True,
    )

    username: Mapped[str] = mapped_column(
        String(50),
        nullable=False,
    )

现在 SQLAlchemy 描述的数据库结构就是：

1
2
3

users
├── id
└── username

接下来需要配置 Alembic。

打开：

1	alembic/env.py

你会看到：

1	target_metadata = None

需要改成：

1
2
3

from app.database import Base

target_metadata = Base.metadata

这里有一个容易踩坑的地方。

仅仅导入 Base 通常还不够。

因为 SQLAlchemy 的模型必须先被 Python 导入，才会注册到：

1	Base.metadata

因此还需要：

1	from app.models.user import User

最终：

from app.database import Base
from app.models.user import User

target_metadata = Base.metadata

虽然这里没有直接使用 User，但导入之后：

User Model
    ↓
注册到 Base.metadata
    ↓
Alembic 才能发现 users 表

如果以后模型越来越多，可以创建：

1	app/models/__init__.py

统一导入：

1
2
3

from .user import User
from .article import Article
from .comment import Comment

然后 env.py 中：

from app.database import Base
import app.models

target_metadata = Base.metadata

这样维护起来会舒服很多。

数据库地址默认在：

1	alembic.ini

里面：

1	sqlalchemy.url = driver://user:pass@localhost/dbname

SQLite：

1	sqlalchemy.url = sqlite:///./app.db

PostgreSQL：

1	sqlalchemy.url = postgresql+psycopg://postgres:password@localhost:5432/mydb

MySQL：

1	sqlalchemy.url = mysql+pymysql://root:password@localhost:3306/mydb

不过实际项目中，不推荐直接把密码写进：

1	alembic.ini

更好的方式是使用 .env：

1	DATABASE_URL=postgresql+psycopg://postgres:password@localhost:5432/mydb

然后通过项目配置读取：

from app.config import settings

config.set_main_option(
    "sqlalchemy.url",
    settings.database_url,
)

这样数据库用户名、密码等敏感信息不会直接提交到 Git。

四、创建和执行第一次数据库迁移

配置完成之后，就可以开始真正使用 Alembic 了。

执行：

1	alembic revision --autogenerate -m "create users table"

这里：

revision

表示创建一个新的数据库版本。

1	--autogenerate

表示让 Alembic 自动比较：

1
2
3

SQLAlchemy Model
        VS
当前数据库 Schema

然后生成差异。

-m

则是给当前 Migration 添加说明。

执行完成后：

1
2
3

alembic/
└── versions/
    └── xxxxx_create_users_table.py

打开这个文件，通常会看到：

def upgrade() -> None:
    op.create_table(
        "users",
        sa.Column(
            "id",
            sa.Integer(),
            nullable=False,
        ),
        sa.Column(
            "username",
            sa.String(length=50),
            nullable=False,
        ),
        sa.PrimaryKeyConstraint("id"),
    )


def downgrade() -> None:
    op.drop_table("users")

这里一定要理解一个区别：

1	alembic revision --autogenerate

只是：

生成数据库迁移文件。

它并不会真正修改数据库。

真正执行迁移需要：

1	alembic upgrade head

其中：

upgrade

表示执行升级。

head

表示升级到当前最新版本。

执行之后，数据库中就会真正出现：

users

表。

与此同时，你还会发现数据库中多了一个：

1	alembic_version

表。

这个表通常只有：

1	version_num

字段，例如：

1
2
3

version_num
----------------
5af233abc81d

Alembic 就是通过它知道：

当前数据库已经执行到了哪个 Migration。

假设迁移历史：

A
↓
B
↓
C
↓
D

数据库中：

1	alembic_version = C

这时候执行：

1	alembic upgrade head

Alembic 就知道：

A 已执行
B 已执行
C 已执行
D 未执行

因此只需要执行：

C → D

五、日常数据库修改应该怎么做

平时使用最多的是这一套流程：

修改 SQLAlchemy Model
        ↓
生成 Migration
        ↓
人工检查 Migration
        ↓
执行 Migration
        ↓
测试
        ↓
提交 Git

比如现在给 User 增加邮箱：

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(
        primary_key=True
    )

    username: Mapped[str] = mapped_column(
        String(50),
        nullable=False,
    )

    email: Mapped[str | None] = mapped_column(
        String(100),
        nullable=True,
    )

生成迁移：

1	alembic revision --autogenerate -m "add email to users"

Alembic 可能生成：

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


def downgrade():
    op.drop_column(
        "users",
        "email",
    )

然后执行：

1	alembic upgrade head

数据库结构变成：

users
├── id
├── username
└── email

Alembic 常见的结构操作基本都可以通过 op 完成。

新增字段：

op.add_column(
    "users",
    sa.Column(
        "email",
        sa.String(length=100),
        nullable=True,
    ),
)

删除字段：

op.drop_column(
    "users",
    "email",
)

创建索引：

op.create_index(
    "ix_users_email",
    "users",
    ["email"],
    unique=True,
)

删除索引：

op.drop_index(
    "ix_users_email",
    table_name="users",
)

修改字段类型：

op.alter_column(
    "users",
    "username",
    existing_type=sa.String(length=50),
    type_=sa.String(length=100),
)

修改字段是否允许 NULL：

op.alter_column(
    "users",
    "email",
    existing_type=sa.String(length=100),
    nullable=False,
)

重命名字段：

op.alter_column(
    "users",
    "username",
    new_column_name="name",
)

修改表名：

op.rename_table(
    "users",
    "accounts",
)

这里有一条原则：

--autogenerate 生成出来的 Migration 一定要人工检查。

因为 Alembic 并不能理解所有业务语义。

例如你把：

username

改成：

name

Alembic 可能无法判断这是字段重命名。

它有可能生成：

op.drop_column(
    "users",
    "username",
)

op.add_column(
    "users",
    sa.Column(
        "name",
        sa.String(50),
    ),
)

如果直接执行：

1	alembic upgrade head

旧的 username 数据可能全部丢失。

而真正应该做的是：

op.alter_column(
    "users",
    "username",
    new_column_name="name",
)

所以不要形成：

生成 Migration
↓
完全不看
↓
直接 upgrade

这种习惯。

六、数据迁移、NOT NULL 和回滚

Alembic 不只能修改数据库结构，也可以做数据迁移，实际项目中经常用到。

假设现在数据库中已经存在大量用户，需要新增：

email

而最终要求：

1	email NOT NULL

如果直接：

op.add_column(
    "users",
    sa.Column(
        "email",
        sa.String(100),
        nullable=False,
    ),
)

数据库中已有的数据没有 email，迁移很可能直接失败。

更加合理的方式是分三步。

先新增一个允许 NULL 的字段：

op.add_column(
    "users",
    sa.Column(
        "email",
        sa.String(100),
        nullable=True,
    ),
)

然后给历史数据补值：

op.execute(
    """
    UPDATE users
    SET email = username || '@example.com'
    WHERE email IS NULL
    """
)

最后修改：

op.alter_column(
    "users",
    "email",
    nullable=False,
)

整个流程就是：

新增 nullable 字段
        ↓
处理历史数据
        ↓
设置 NOT NULL

这种操作属于：

1
2
3

Schema Migration
+
Data Migration

Alembic 中可以直接使用：

1	op.execute(...)

运行 SQL，例如：

op.execute(
    """
    UPDATE users
    SET status = 'active'
    WHERE status IS NULL
    """
)

因此它也适合做：

历史数据补充
数据格式转换
枚举值转换
旧字段迁移
数据清理

数据库升级之后，也可以进行回滚。

回滚一个版本：

1	alembic downgrade -1

例如：

V1
↓
V2
↓
V3

现在数据库在：

V3

执行：

1	alembic downgrade -1

会回到：

V2

继续执行：

1	alembic downgrade -1

回到：

V1

也可以直接回滚到指定版本：

1	alembic downgrade <revision_id>

甚至回滚全部：

1	alembic downgrade base

开发环境中可能会使用这个命令，但生产环境一定要谨慎。

常用版本查看命令：

1	alembic current

查看当前数据库版本。

1	alembic history

查看迁移历史。

1	alembic history --verbose

查看更加详细的历史。

1	alembic heads

查看当前最新 Migration。

七、FastAPI、异步 SQLAlchemy 和真实项目结构

一个比较常见的 FastAPI 项目结构可以设计成：

backend/
├── alembic/
│   ├── versions/
│   ├── env.py
│   └── script.py.mako
│
├── app/
│   ├── api/
│   │   └── routes/
│   │
│   ├── core/
│   │   └── config.py
│   │
│   ├── db/
│   │   ├── base.py
│   │   └── session.py
│   │
│   ├── models/
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── article.py
│   │
│   ├── schemas/
│   └── main.py
│
├── .env
├── alembic.ini
└── pyproject.toml

注意：

models

和：

schemas

不是一个东西。

models：

1	SQLAlchemy 数据库模型

schemas：

1	Pydantic 请求 / 响应数据模型

这是 FastAPI 项目非常常见的一种分层方式。

现在很多 FastAPI 项目也会使用异步 SQLAlchemy，例如：

from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(
    DATABASE_URL
)

PostgreSQL URL：

1	DATABASE_URL=postgresql+asyncpg://postgres:password@localhost:5432/demo

安装：

1	uv add asyncpg

如果你的项目从一开始就准备使用异步 SQLAlchemy，可以直接初始化 Alembic 的 async 模板：

1	alembic init -t async alembic

这样生成出来的 env.py 会自动包含异步迁移结构。

核心逻辑类似：

async def run_async_migrations():
    async with connectable.connect() as connection:
        await connection.run_sync(
            do_run_migrations
        )

最终通过：

1
2
3

asyncio.run(
    run_async_migrations()
)

运行。

对于 FastAPI 异步项目，使用 async 模板比自己手动修改普通 env.py 更方便。

八、多人协作和生产环境容易踩的坑

多人协作时，Alembic 会遇到一些新问题。

最典型的就是多个 Migration 分支。

假设：

main

上有一个 Migration。

开发者 A 创建：

1
2
3

main
 ↓
A

开发者 B 同时创建：

1
2
3

main
 ↓
B

两个分支合并之后，就变成：

      A
     /
main
     \
      B

这时候运行：

1	alembic heads

可能会看到两个 Head。

可以通过：

1	alembic merge heads -m "merge migrations"

创建一个新的 Merge Migration：

      A
     / \
main    M
     \ /
      B

然后：

1	alembic upgrade head

继续迁移。

还有一条原则：

已经进入共享环境的 Migration，尽量不要删除或者随意修改。

比如当前版本历史：

A
↓
B
↓
C

如果 B 已经被：

1
2
3

其他开发人员
测试服务器
生产服务器

执行过了。

后来发现 B 有问题，不应该直接删除 B。

更加合理的是：

A
↓
B
↓
C
↓
D 修复 B

可以把已经共享出去的 Migration 理解成：

Git 中已经发布出去的历史 Commit。

尽量通过新的 Migration 修复，而不是直接篡改历史。

如果一个数据库本来就已经存在表，只是后来才接入 Alembic，还可能会使用：

1	alembic stamp head

stamp 的意思是：

不执行 Migration，只修改 Alembic 记录的数据库版本。

比如现有数据库已经手动拥有：

1
2
3

users
articles
comments

但没有：

1	alembic_version

如果确定数据库结构和当前 Migration Head 一致，可以：

1	alembic stamp head

告诉 Alembic：

1	当前数据库已经处于 head

需要注意：

stamp

不会创建表、删除表或修改字段。

它只是修改版本记录。

另外还有一种危险的情况。

执行：

1	alembic revision --autogenerate -m "update"

结果突然生成：

1
2
3

op.drop_table("users")
op.drop_table("articles")
op.drop_table("comments")

这时候不要直接执行。

通常说明：

1	target_metadata 配置错误

或者：

1	SQLAlchemy Models 没有被导入

例如数据库里面有：

users

但：

1	Base.metadata

里面没有 users。

Alembic 就会认为：

Model 中已经删除了 users，所以数据库也应该删除。

可以临时检查：

1	print(Base.metadata.tables.keys())

正常情况应该看到：

dict_keys([
    "users",
    "articles",
    "comments"
])

如果：

1	dict_keys([])

说明 Model 根本没有注册成功。

九、部署时应该怎么使用 Alembic

Migration 文件属于项目代码的一部分，所以：

1 2	alembic/ └── versions/

必须提交 Git。

例如：

git add alembic
git add alembic.ini

git commit -m "feat: add user email migration"

服务器部署新版本时，一般流程是：

拉取新代码
↓
安装依赖
↓
执行数据库 Migration
↓
启动新版本应用

对应：

1
2
3

git pull
uv sync
alembic upgrade head

然后：

1	uv run uvicorn app.main:app

如果使用 Docker，简单项目有时候会这样：

services:
  backend:
    build: .
    command: >
      sh -c "
      alembic upgrade head &&
      uvicorn app.main:app --host 0.0.0.0 --port 8000
      "

即：

Docker 容器启动
↓
执行 Migration
↓
启动 FastAPI

但大型生产环境一般不推荐让所有 Web 容器都同时运行 Migration。

因为如果一次启动多个实例：

1
2
3

Backend 1
Backend 2
Backend 3

都同时：

1	alembic upgrade head

可能产生竞争。

更规范的方式是：

独立执行 Migration
↓
Migration 成功
↓
部署应用实例

在 CI/CD 中通常也会单独设置数据库迁移阶段。

十、常用命令和最终理解

日常开发常用的 Alembic 命令不多。

初始化：

1	alembic init alembic

如果使用异步 SQLAlchemy：

1	alembic init -t async alembic

创建空 Migration：

1	alembic revision -m "message"

自动生成 Migration：

1	alembic revision --autogenerate -m "message"

升级到最新版：

1	alembic upgrade head

升级一个版本：

1	alembic upgrade +1

回滚一个版本：

1	alembic downgrade -1

查看当前版本：

1	alembic current

查看迁移历史：

1	alembic history

查看 Head：

1	alembic heads

标记版本：

1	alembic stamp head

合并多个 Head：

1	alembic merge heads -m "merge heads"

开发项目时，要养成的习惯是：

修改 Model
    ↓
revision --autogenerate
    ↓
人工检查 Migration
    ↓
upgrade head
    ↓
测试 downgrade
    ↓
再次 upgrade
    ↓
提交 Git

一句话概括：

Alembic 是 SQLAlchemy 生态中的数据库版本控制工具。

Git 管理：

代码版本

Alembic 管理：

1	数据库 Schema 版本

例如：

Git

Commit A
↓
Commit B
↓
Commit C

对应数据库：

Alembic

Migration A
↓
Migration B
↓
Migration C

所以一个后端项目同时有两份历史：

1
2
3

代码版本历史
+
数据库版本历史

对于 FastAPI 项目来说，常见的组合是：

FastAPI
    +
SQLAlchemy
    +
Alembic
    +
PostgreSQL

刚开始学习 Alembic 时，不需要一次记住所有 API。

优先掌握：

alembic revision --autogenerate -m "xxx"

alembic upgrade head

alembic downgrade -1

alembic current

alembic history

然后理解：

Model
↓
Metadata
↓
Migration
↓
Database

之间的关系。