基于Alembic的python数据库迁移笔记|violet blog

数据库迁移指南（Alembic 使用说明）

项目代码地址：https://github.com/carolin-violet/violet-wallpaper-backend

本项目使用 Alembic + SQLAlchemy 管理数据库结构变更，结合 uv 命令和项目内的配置文件，实现：

模型变更与数据库结构的同步（alembic revision --autogenerate + alembic upgrade）
异步数据库引擎支持
统一的命名规范与迁移脚本风格（集成 Ruff 格式化和检查）
数据初始化（通过 src/alembic/data 下的工具和 JSON 文件）

本文结合以下文件进行说明：

alembic.ini
src/alembic/env.py
src/alembic/script.py.mako
pyproject.toml
src/database.py

一、整体架构与配置入口

Alembic 主配置：alembic.ini
- script_location = src/alembic：迁移脚本所在目录指向项目内 src/alembic。
- file_template = %(year)d-%(month).2d-%(day).2d_%(rev)s_%(slug)s：迁移文件名自动带日期前缀，便于按时间管理。
- prepend_sys_path = .：将项目根目录加入 sys.path，支持 import src.xxx 风格。
- sqlalchemy.url = ...：默认数据库 URL，这个值会在 src/alembic/env.py 中被 settings.DATABASE_URL 覆盖。

数据库基础设施：src/database.py

engine = create_async_engine(
    settings.DATABASE_URL,
    echo=settings.DEBUG,
    future=True,
    pool_pre_ping=True,
)
async_session_maker = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

class Base(DeclarativeBase):
    metadata = MetaData(
        naming_convention={
            "ix": "%(column_0_label)s_idx",
            "uq": "%(table_name)s_%(column_0_name)s_key",
            "ck": "%(table_name)s_%(constraint_name)s_check",
            "fk": "%(table_name)s_%(column_0_name)s_fkey",
            "pk": "%(table_name)s_pkey",
        }
    )

使用 异步引擎（create_async_engine）和 异步会话。
通过 settings.DATABASE_URL 统一数据库连接配置。
使用统一的 naming_convention，让索引/约束名称在迁移中可预测，有利于 Alembic 生成和维护。

Alembic 环境脚本：src/alembic/env.py
核心部分：
```
import src.models.dictionary  # noqa: F401
import src.models.picture  # noqa: F401
import src.models.tag  # noqa: F401
from alembic import context
from src.conf import settings
from src.database import Base

config = context.config
config.set_main_option("sqlalchemy.url", settings.DATABASE_URL)
target_metadata = Base.metadata
```
- 显式导入所有模型（dictionary / picture / tag），确保 自动迁移 能扫到完整的 Base.metadata。
- 使用 settings.DATABASE_URL 覆盖 alembic.ini 中的 sqlalchemy.url，保证和应用代码使用同一套配置。
- target_metadata = Base.metadata：Alembic 以 Base 作为模型元数据来源。
- 使用 async_engine_from_config + asyncio.run(...) 实现 在线迁移的异步引擎 支持。

二、迁移脚本模板与代码风格

迁移模板：src/alembic/script.py.mako

from alembic import op
import sqlalchemy as sa
# 数据初始化工具（可选使用）
# from src.alembic.data.init_utils import init_dictionary_data, init_table_data, load_json_data

# revision identifiers, used by Alembic.
revision = ...
down_revision = ...

def upgrade() -> None:
    ...

def downgrade() -> None:
    ...

特点：

默认生成 带类型标注 的 upgrade() -> None / downgrade() -> None。
预留了数据初始化工具的导入注释，方便在迁移中调用通用的数据初始化逻辑。

迁移文件自动格式化与检查：alembic.ini + pyproject.toml
在 alembic.ini 中配置了 post_write_hooks：
```
[post_write_hooks]
hooks = ruff-format, ruff-check
ruff-format.type = exec
ruff-format.executable = uvx
ruff-format.options = ruff format REVISION_SCRIPT_FILENAME
ruff-check.type = exec
ruff-check.executable = uvx
ruff-check.options = ruff check --fix REVISION_SCRIPT_FILENAME
```
作用：
- 每次执行 alembic revision 生成迁移脚本后，自动执行：
  - uvx ruff format ...：统一格式化
  - uvx ruff check --fix ...：静态检查并自动修复部分问题
- 配合 pyproject.toml 中的 Ruff 配置（行宽、规则集等），保证迁移文件风格与项目代码一致。
为了让 ruff 可用，项目在 pyproject.toml 的 dev 组中添加了：
```
[dependency-groups]
dev = [
    "pytest>=8.4.2,<9",
    "pytest-asyncio>=1.3.0",
    "pytest-cov>=7.0.0",
    "pytest-mock>=3.15.1",
    "ruff>=0.14.9",
    "testcontainers[postgres]>=4.10.0",
]
```
建议：开发环境中执行一次 uv sync --group dev，确保 ruff 等开发依赖已安装。

三、常用命令与典型场景

1. 初始化 / 同步数据库

前提：PostgreSQL 服务已启动，.env 中的 DATABASE_URL 已配置并与目标库匹配。

首次初始化数据库结构：

# 将数据库升级到最新版本（head）
uv run alembic upgrade head

查看迁移历史与当前版本：

# 查看所有迁移版本
uv run alembic history

# 查看当前数据库对应的迁移版本
uv run alembic current

2. 修改模型并生成迁移

工作流程（推荐）：

修改 src/models/ 下的模型（如 picture.py、dictionary.py、tag.py）。

生成自动迁移脚本：

uv run alembic revision --autogenerate -m "描述你的更改"

打开 src/alembic/versions/ 下新生成的迁移文件，确认：
- 是否只包含预期的变更（新增字段/表、修改类型等）。
- 自动生成的 upgrade/downgrade 逻辑是否合理。
应用迁移：
```
uv run alembic upgrade head
```

注意：

如果自动生成的迁移过于复杂（例如涉及数据迁移、拆表合表），可以用：
```
uv run alembic revision -m "描述你的更改"
```
生成一个空迁移文件，再手工编写 upgrade / downgrade。

3. 回滚 / 调试迁移

回退一个版本：
```
uv run alembic downgrade -1
```
回退到指定版本：
```
uv run alembic downgrade <revision_id>
```

查看不执行的 SQL 语句（仅打印 SQL，不真正执行）：

# 查看升级到 head 的 SQL
uv run alembic upgrade head --sql

# 查看回退一个版本的 SQL
uv run alembic downgrade -1 --sql

四、数据初始化与幂等性

项目在 src/alembic/data 下提供了数据初始化工具（如 init_utils.py），并在迁移模板中预留导入注释：

# from src.alembic.data.init_utils import init_dictionary_data, init_table_data, load_json_data

典型用法示例（伪代码，仅说明思路）：

def upgrade() -> None:
    # 创建表结构 ...

    # 从 JSON 文件初始化字典数据，并基于某个字段做去重检查
    init_dictionary_data(
        table_name="dictionary",
        data_file="dictionary_categories.json",
        check_column="code",
        default_values={"type": -1},
    )

设计原则：

幂等性：同一迁移多次执行不会产生重复数据，常用方式是：
- check_column 指定唯一业务键（如 code），插入前先查是否存在。
- 对缺失字段（如 type）通过 default_values 自动填充。
职责分离：结构变更（DDL）与数据初始化（DML）都在同一迁移中，但具体数据从 JSON 文件读取，便于维护和复用。

五、与应用代码的协同关系

统一的连接配置
- 应用代码和 Alembic 都依赖 settings.DATABASE_URL：
  - src/database.py 中用它创建异步引擎。
  - src/alembic/env.py 中用 config.set_main_option("sqlalchemy.url", settings.DATABASE_URL) 覆盖 Alembic 的连接配置。
- 好处：只需在 .env 中维护一份数据库配置，避免环境不一致。
模型导入顺序与 Base.metadata
- env.py 中显式导入 src.models.dictionary/picture/tag，并设置：
```
target_metadata = Base.metadata
```
- 要让 Alembic 正确感知新的模型或字段，新增模型时务必确保被某个导入路径引用：
  - 推荐在 src/models/__init__.py 中统一导入/导出所有模型，然后在 env.py 中只导入 src.models。
命名规范与迁移一致性
- Base.metadata 的 naming_convention 会影响 Alembic 生成的索引/约束名，如果你手工写迁移语句，应遵循同一规范，避免名称冲突。

六、常见问题与注意事项

数据库结构与迁移记录不一致
场景：手动删除了某些表，或者直接改了数据库结构，导致执行 alembic upgrade 报错（例如 “Target database is not up to date” 或 “表不存在但迁移版本已记录”）。
处理方式（README 中也有记录）：
```
# 1. 重置迁移历史到基础版本（仅修改 alembic_version 表，不执行 DDL）
uv run alembic stamp base

# 2. 重新执行所有迁移
uv run alembic upgrade head
```
注意：
- 这会从逻辑上“认为”数据库是空白状态，然后重新跑一遍所有迁移。
- 如果迁移中包含数据初始化逻辑，必须确保它们是幂等的。
异步引擎与 Alembic 的关系
- 运行时使用的是异步引擎（AsyncSession），但 迁移脚本内部通常还是写同步的 SQLAlchemy 操作（通过 Alembic 的 op 对象）。
- env.py 中使用 async_engine_from_config 和 run_sync(...)，Alembic 自己会在内部桥接异步连接和同步迁移逻辑。
开发/生产环境数据库切换
- 通过 .env 中的 DATABASE_URL 区分不同环境（例如本地、测试、生产）。
- 迁移命令本身不区分环境，连接哪个库完全由当前环境变量决定，执行前务必确认当前 DATABASE_URL 指向的是目标环境。
Ruff 相关问题
- 如果执行 alembic revision 时提示找不到 ruff，请先安装 dev 依赖：
```
uv sync --group dev
```
- 或在 CI 环境中显式安装 ruff，以保证 post_write_hooks 正常执行。

七、推荐开发流程总结

修改或新增模型（src/models/*.py）。

运行：

uv run alembic revision --autogenerate -m "描述你的更改"

检查 src/alembic/versions/ 中新生成的迁移文件，必要时手工调整。
确认无误后：
```
uv run alembic upgrade head
```
如需回滚，使用：
```
uv run alembic downgrade -1
```
对涉及数据初始化的迁移，优先使用 src/alembic/data 中的工具函数，并保证逻辑幂等。