4. Alembic 入门：安装、初始化与异步适配

Alembic 是 SQLAlchemy 官方推荐的数据库迁移工具，用于管理数据库 Schema 的版本变更。在异步中间件集成场景中，Alembic 需要配合异步引擎和异步 Session 工作。本节从零开始，完成 Alembic 的安装、项目初始化、异步适配以及第一个迁移脚本的生成。

4.1 安装 Alembic

使用 pip 安装 Alembic，建议在虚拟环境中操作：

pip install alembic

验证安装版本：

alembic --version
# 输出示例：alembic 1.13.1

4.2 初始化 Alembic 项目

在项目根目录执行初始化命令：

alembic init alembic

初始化后生成的标准目录结构如下：

文件/目录	用途
alembic/	迁移脚本存放目录
alembic/versions/	版本迁移脚本（按时间戳命名）
alembic/env.py	迁移环境配置，核心适配文件
alembic.ini	全局配置文件，含数据库连接串
alembic/script.py.mako	迁移脚本模板

4.3 异步适配 env.py

默认生成的 env.py 使用同步方式运行。在异步中间件项目中，必须改造为异步上下文。核心改动点包括：

• 使用 async_engine 替代同步 create_engine
• 使用 AsyncConnection 运行迁移
• 使用 run_async() 包装异步入口

以下是完整的异步适配 env.py 示例：

import asyncio
from logging.config import fileConfig

from alembic import context
from sqlalchemy.ext.asyncio import create_async_engine, AsyncConnection

# 导入你的 SQLAlchemy Base 和模型
from app.database import Base
from app.models import User, Task  # 示例模型

config = context.config
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

target_metadata = Base.metadata

# 从 alembic.ini 读取异步连接串
# 注意：连接串前缀应为 "postgresql+asyncpg://" 或 "mysql+aiomysql://"
DATABASE_URL = config.get_main_option("sqlalchemy.url")

def run_migrations_offline() -> None:
    """离线模式：生成 SQL 脚本，不连接数据库"""
    context.configure(
        url=DATABASE_URL,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )
    with context.begin_transaction():
        context.run_migrations()

def do_run_migrations(connection: AsyncConnection) -> None:
    """在异步连接上执行迁移"""
    context.configure(connection=connection, target_metadata=target_metadata)
    with context.begin_transaction():
        context.run_migrations()

async def run_async_migrations() -> None:
    """异步入口：创建异步引擎并执行迁移"""
    connectable = create_async_engine(DATABASE_URL, echo=True)
    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)
    await connectable.dispose()

def run_migrations_online() -> None:
    """在线模式入口：启动异步事件循环"""
    asyncio.run(run_async_migrations())

if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

4.4 异步适配架构图

下图展示了 Alembic 异步迁移的核心调用链路：

4.5 生成第一个迁移脚本

完成异步适配后，使用 Alembic 自动检测模型变更生成迁移脚本：

alembic revision --autogenerate -m "create_user_and_task_tables"

执行成功后，在 alembic/versions/ 目录下生成一个以时间戳命名的 Python 文件，内容结构如下：

"""create_user_and_task_tables

Revision ID: a1b2c3d4e5f6
Revises: 
Create Date: 2025-03-21 14:30:00.123456
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa

# revision identifiers
revision: str = 'a1b2c3d4e5f6'
down_revision: Union[str, None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None

def upgrade() -> None:
    # 自动生成的升级操作
    op.create_table('users',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('username', sa.String(length=50), nullable=False),
        sa.Column('email', sa.String(length=120), nullable=False),
        sa.PrimaryKeyConstraint('id'),
        sa.UniqueConstraint('email')
    )
    op.create_table('tasks',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('title', sa.String(length=200), nullable=False),
        sa.Column('user_id', sa.Integer(), nullable=False),
        sa.ForeignKeyConstraint(['user_id'], ['users.id'], ),
        sa.PrimaryKeyConstraint('id')
    )

def downgrade() -> None:
    # 自动生成的回滚操作
    op.drop_table('tasks')
    op.drop_table('users')

4.6 迁移脚本生成流程

下图展示了从模型定义到迁移脚本生成的完整流程：

4.7 执行迁移

生成迁移脚本后，执行以下命令将变更应用到数据库：

alembic upgrade head

常用迁移命令速查表：

命令	作用
alembic upgrade head	升级到最新版本
alembic downgrade -1	回退一个版本
alembic history	查看迁移历史
alembic current	查看当前数据库版本
alembic revision --autogenerate -m "msg"	自动生成迁移脚本

4.8 常见问题与避坑

• 异步连接串错误： 使用 postgresql+asyncpg:// 而非 postgresql://，否则 create_async_engine 会报错。
• env.py 未正确导入模型： 确保 target_metadata = Base.metadata 之前已导入所有模型类，否则自动检测不到表变更。
• 迁移脚本生成空文件： 检查 alembic.ini 中的 sqlalchemy.url 是否指向正确的数据库，以及模型是否已注册到 Base。
• 异步事件循环冲突： 如果项目中已有运行中的事件循环，使用 asyncio.run() 可能引发冲突。可改用 loop.run_until_complete() 或使用 alembic 的异步上下文管理器。