2、常见异步迁移框架：Alembic异步模式、SQLAlchemy async session、自定义异步迁移脚本

在异步化开发阶段，数据库迁移不再只是同步脚本的简单封装。异步迁移框架需要正确处理事件循环、连接池生命周期以及异步会话的上下文管理。本节聚焦三种主流实践：Alembic 原生异步模式、SQLAlchemy async session 的迁移适配以及自定义异步迁移脚本的编写与故障排查。

2.1 三种异步迁移方案对比

2.2 Alembic 异步模式：run_async 的正确用法

Alembic 从 1.7+ 开始支持异步迁移。核心思路是在 env.py 中定义 run_migrations_online() 时，使用 asyncio.run() 调用异步上下文。故障高发区在于 事件循环嵌套 和 连接池未显式释放。

# alembic/env.py
from alembic import context
from sqlalchemy.ext.asyncio import create_async_engine, AsyncConnection

async def run_migrations_online():
    connectable = create_async_engine(
        config.get_main_option("sqlalchemy.url"),
        pool_size=5,
        max_overflow=10,
    )

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

    await connectable.dispose()  # 必须显式释放连接池

def do_run_migrations(connection):
    context.configure(connection=connection, target_metadata=target_metadata)
    with context.begin_transaction():
        context.run_migrations()

# 入口
if context.is_offline_mode():
    context.run_migrations()
else:
    asyncio.run(run_migrations_online())

2.3 SQLAlchemy async session 迁移适配

当迁移脚本需要执行复杂数据操作（如批量更新、数据校验）时，直接使用 AsyncSession 比 Alembic 的 run_sync 更灵活。但必须注意：迁移脚本中的每个数据库操作都必须是异步的，不能混用同步 Session。

# 迁移脚本示例：使用 AsyncSession 执行数据迁移
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from sqlalchemy import text

async def upgrade_data(async_session: AsyncSession):
    # 异步执行复杂数据操作
    result = await async_session.execute(
        text("UPDATE users SET status = 'active' WHERE last_login > NOW() - INTERVAL '30 days'")
    )
    await async_session.commit()
    print(f"Updated {result.rowcount} rows")

def upgrade():
    # 在 Alembic 迁移中调用
    from alembic import op
    from sqlalchemy.ext.asyncio import create_async_engine

    engine = create_async_engine(op.get_context().config.get_main_option("sqlalchemy.url"))
    async_session_factory = async_sessionmaker(engine, class_=AsyncSession)

    async def run():
        async with async_session_factory() as session:
            await upgrade_data(session)

    import asyncio
    asyncio.run(run())

2.4 自定义异步迁移脚本架构

对于超出 Alembic 能力范围的迁移（如跨数据库、分片集群、调用外部 API 进行数据清洗），需要编写独立的异步迁移脚本。这类脚本的故障排查重点在于 任务编排 和 异常传播。

2.5 自定义脚本故障排查清单

• 事件循环冲突：确保整个迁移过程只有一个事件循环。不要在 async def 内部调用 asyncio.run()，改用 await 或 asyncio.create_task()。
• 连接池泄漏：每个 async with engine.connect() 必须配对 await engine.dispose()。使用 async with 上下文管理器自动释放。
• 事务边界模糊：自定义脚本中每个独立迁移步骤应包裹在单独的事务中，避免一个步骤失败导致整个迁移回滚。
• 日志缺失：异步异常栈可能被事件循环吞没。在 main() 入口添加 try/except 并输出完整 traceback。
• 幂等性缺失：迁移脚本应支持重复执行。使用 IF NOT EXISTS、CREATE OR REPLACE 或先检查再操作。

2.6 三种方案的选型建议

根据项目阶段和团队异步经验，推荐以下选型路径：

1. 已有同步迁移项目，逐步异步化 → 优先使用 Alembic 异步模式，改动最小，风险可控。
2. 新建异步项目，全栈异步 → 使用 SQLAlchemy async session 配合 Alembic，保持代码风格一致。
3. 复杂数据迁移、跨库操作、分片迁移 → 使用 自定义异步迁移脚本，但必须严格遵循任务编排和异常传播规范。