1. 迁移

ORM 模型只是 Python 对数据库结构的描述。修改 Task 类不会自动改变已经运行的 PostgreSQL 表，也不应该让应用启动时自行猜测并改写生产结构。

数据库迁移把每次结构变化写进版本文件。每个文件都有自己的 revision，并通过 down_revision 指向父版本，因此迁移历史实际是一张版本图。大多数项目起初只有一条线性路径，看起来类似下面这样：

1
空数据库
2
  -> 7ac21f 创建 users
3
  -> 12bd90 创建 projects 和 tasks
4
  -> 3e8a41 增加 tasks.version
5
  -> 91fc72 创建任务列表索引

后面的命令会反复使用几个迁移术语，可以先建立对应关系：

例如数据库的 current 是 12bd90，代码中的 head 是 91fc72，执行 upgrade head 时，Alembic 会沿版本图依次运行中间两个 revision 的 upgrade()。它不是每次都重新执行全部迁移。

同一套迁移文件会与应用代码一起进入 Git。测试、预发布和生产环境从各自的当前版本出发，按相同的版本图执行到目标版本，数据库结构才会与应用代码保持一致。

Alembic 是 SQLAlchemy 生态中的迁移工具。当前项目使用 pyproject.toml 管理 Python 依赖，并通过异步数据库驱动连接 PostgreSQL，因此可以这样初始化：

1
uv add "alembic>=1.19,<2"
2
uv run alembic init --template pyproject_async migrations

初始化完成后，项目中会出现以下配置和目录：

初始化完成时，versions/ 还是空目录。执行后文的 revision 命令后，迁移文件才会出现在这里。

pyproject_async 模板会把迁移目录等工具配置追加到 pyproject.toml。对当前目录结构而言，最关键的配置如下：

1
[tool.alembic]
2
script_location = "%(here)s/migrations"
3
prepend_sys_path = ["."]

%(here)s 表示 pyproject.toml 所在目录，prepend_sys_path 让 Alembic 从项目根目录导入 app 包。后续命令都应在项目根目录执行，否则 .env、应用包和相对路径可能无法按预期找到。

alembic.ini 默认保留数据库 URL 占位项和日志配置。真实密码不要写入 pyproject.toml 或 alembic.ini，后面会在 env.py 中从环境配置读取连接地址，并覆盖占位 URL。

异步模板只改变 Alembic 建立数据库连接的方式。迁移文件中的 upgrade()、downgrade() 和 op.* 仍然是同步函数，Alembic 会通过 SQLAlchemy 的 run_sync() 在异步连接上运行它们。

2. 迁移环境

自动生成迁移前，Alembic 必须取得完整的目标元数据。项目应在 app/db/models/__init__.py 中显式导入全部模型，这样 import app.db.models 才能执行每个模型类的定义，并把表注册到 Base.metadata。如果 __init__.py 没有导入子模块，只导入包名并不能完成注册。

将 migrations/env.py 调整为下面的完整结构：

01
import asyncio
02
from logging.config import fileConfig
03

04
from alembic import context
05
from sqlalchemy import pool
06
from sqlalchemy.engine import Connection
07
from sqlalchemy.ext.asyncio import async_engine_from_config
08

09
from app.core.settings import settings
10
from app.db.base import Base
11
import app.db.models  # noqa: F401
12

13

14
config = context.config
15

16
if config.config_file_name is not None:
17
    fileConfig(config.config_file_name)
18

19
config.set_main_option(
20
    "sqlalchemy.url",
21
    settings.database_url.replace("%", "%%"),
22
)
23
target_metadata = Base.metadata
24

25

26
def run_migrations_offline() -> None:
27
    url = config.get_main_option("sqlalchemy.url")
28
    if url is None:
29
        raise RuntimeError("sqlalchemy.url is not configured")
30

31
    context.configure(
32
        url=url,
33
        target_metadata=target_metadata,
34
        literal_binds=True,
35
        dialect_opts={"paramstyle": "named"},
36
        compare_type=True,
37
        compare_server_default=True,
38
        transaction_per_migration=True,
39
    )
40

41
    with context.begin_transaction():
42
        context.run_migrations()
43

44

45
def do_run_migrations(connection: Connection) -> None:
46
    context.configure(
47
        connection=connection,
48
        target_metadata=target_metadata,
49
        compare_type=True,
50
        compare_server_default=True,
51
        transaction_per_migration=True,
52
    )
53

54
    with context.begin_transaction():
55
        context.run_migrations()
56

57

58
async def run_async_migrations() -> None:
59
    connectable = async_engine_from_config(
60
        config.get_section(config.config_ini_section, {}),
61
        prefix="sqlalchemy.",
62
        poolclass=pool.NullPool,
63
    )
64

65
    async with connectable.connect() as connection:
66
        await connection.run_sync(do_run_migrations)
67

68
    await connectable.dispose()
69

70

71
def run_migrations_online() -> None:
72
    asyncio.run(run_async_migrations())
73

74

75
if context.is_offline_mode():
76
    run_migrations_offline()
77
else:
78
    run_migrations_online()

这段配置包含两条入口。在线模式建立真实连接，用于自动比较和直接升级数据库；离线模式不连接数据库，可以把已有迁移操作渲染成 SQL。NullPool 表示命令执行完就释放连接，不为短生命周期的迁移进程维护连接池。

config.set_main_option() 使用配置插值语法，因此 URL 中的 % 要写成 %%，否则包含百分号编码的密码可能无法解析。转义只发生在 Alembic 配置层，不会改变最终传给数据库驱动的 URL。

这里显式开启类型和服务端默认值比较。服务端默认值的自动比较并非在所有表达式上都完全可靠，PostgreSQL 还可能在数据库中计算表达式以判断它们是否等价，因此生成结果仍然需要人工复核。

transaction_per_migration=True 让每个 revision 使用各自的事务。普通 PostgreSQL DDL 可以随事务提交或回滚，这种边界也更适合后文包含自动提交块的并发索引迁移；不过，单个 revision 内进入自动提交块时，块之前的操作仍然会被提交，所以仍不能把不相关操作混在一起。

需要交给 DBA 审查 SQL 时，可以使用离线模式：

1
uv run alembic upgrade head --sql > migration.sql

这条命令只渲染已经存在的 revision，不会连接并修改数据库。revision --autogenerate 则必须连接数据库，读取当前结构后才能比较，不能用离线模式代替。自定义 Python 数据处理也未必能完整渲染成离线 SQL，因此生成文件仍需人工检查。

3. 建立迁移历史

自动生成迁移时，Alembic 比较的是目标数据库当前结构和应用模型中的 Base.metadata，不是只读取上一份迁移文件：

1
目标数据库当前结构
2
        ↓ 比较
3
应用模型 Base.metadata
4
        ↓
5
候选 upgrade() / downgrade()

因此，命令中的 --autogenerate 只负责生成候选文件，不会执行文件，也不会修改数据库。如果项目从空数据库开始，先确保 Base.metadata 包含所有需要 Alembic 管理的表、列、约束和索引，再生成初始版本：

1
uv run alembic revision --autogenerate -m "create initial schema"
2
uv run alembic upgrade head

第一条命令会比较空数据库与模型元数据，并在 migrations/versions/ 中生成候选文件。必须先检查文件内容，再执行第二条命令。upgrade head 会读取数据库的 current revision，按依赖顺序执行尚未应用的 upgrade()；升级成功后，数据库会创建业务结构和 alembic_version 表。

另一种情况是数据库已经由手写 SQL 建好，现在才开始使用 Alembic。此时不能直接对现有数据库执行一份再次创建相同表的初始迁移。可以先连接临时空数据库生成并审查初始版本，再把配置切回现有数据库，确认现有结构与模型一致：

1
uv run alembic check
2
uv run alembic stamp head

alembic check 使用与自动生成相同的比较机制。它没有报告待生成操作，只能说明该机制没有发现结构差异，不能验证业务数据，也不能弥补自动比较本身的限制。仍然要人工检查列、默认值、约束、索引和扩展对象。

stamp head 不会执行任何 upgrade()，只会把版本号写入 alembic_version。只有确认现有数据库已经等价于该版本时才能使用；把它当作跳过失败迁移的办法，会让版本记录与真实结构分离。