如何在同步Alembic迁移中运行SQLAlchemy上传协程？

我尝试通过Alembic迁移填充数据库，流程为运行异步上传器获取文件，解析为SQLModel后使用SQLAlchemy插入数据库。已将Alembic初始化为异步模式，env文件采用官方模板，当前迁移代码如下：

def upload_func():
    loop = asyncio.new_event_loop()
    try:
        return loop.run_until_complete(upload_coro())
    finally:
        loop.close()

def upgrade():
    with ThreadPoolExecutor(max_workers=1) as executor:
        future = executor.submit(upload_func)
        future.result()

首次迁移运行正常，但后续每次迁移执行到await self.session.execute(stmt)时，都会触发以下错误：

RuntimeError: Task <Task pending coro=<upload_coro() running at /backend/src/alembic/versions/upload_smth.py:32> cb=[_run_until_complete_cb() at /Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/asyncio/base_events.py:157]> got Future attached to a different loop

需要多次执行alembic upgrade head才能完成迁移。想明确以下问题：

• 当前操作的错误点在哪里？
• 如何解决“Future关联不同循环”的报错？
• 在Alembic同步upgrade()函数中运行SQLAlchemy协程的正确方法是什么？

错误根源 #

当前写法存在两个核心问题：

1. 循环上下文冲突：手动创建新事件循环+线程池的组合，会导致异步任务（如数据库会话操作）可能绑定到Alembic自身维护的全局循环，而非你手动创建的循环，引发循环不匹配。
2. 冗余的线程池封装：Alembic异步模式本身已在异步上下文环境中运行，无需额外通过线程池包裹异步代码，反而会破坏循环上下文的一致性。

正确实现方式 #

方案1：复用Alembic的现有异步循环（推荐） #

直接获取Alembic初始化好的运行中循环，执行异步任务：

import asyncio

def upgrade():
    loop = asyncio.get_running_loop()
    loop.run_until_complete(upload_coro())

方案2：兼容老版本Python的循环管理（不推荐） #

若使用Python 3.7以下版本，需先检查并复用已有循环，避免重复创建：

import asyncio

def upgrade():
    try:
        # 优先获取当前运行中的循环
        loop = asyncio.get_running_loop()
    except RuntimeError:
        # 无运行中循环时手动创建并设置为全局循环
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
    try:
        loop.run_until_complete(upload_coro())
    finally:
        # 仅关闭手动创建的循环
        if not loop.is_running():
            loop.close()

关键注意事项 #

• 确保upload_coro中使用的数据库会话，是从Alembic异步env提供的异步引擎创建的，不要在协程内手动新建异步连接/会话，防止会话绑定到错误循环。
• 异步协程中若包含同步阻塞操作（如文件读取），需用loop.run_in_executor包装，避免阻塞异步循环。

内容的提问来源于stack exchange，提问作者flintpnz