如何编写支持异步与同步环境的Python通用辅助函数？

我正尝试为项目编写通用辅助函数，但部分遗留项目处于同步环境中。最初写的辅助函数如下：

def func_for_async_and_sync(session, data):
    statement = select(Obj).where(Obj.id == data['id'])
    # 其他更多代码
    if isinstance(session, AsyncSession):
        obj = await session.execute(statement)
    else:
        obj = session.execute(statement)
    # 修改obj并返回
    return obj

显然这个函数无法正常运行，因为普通函数里不能使用await关键字。我需要在普通函数中调用AsyncSession.execute并获取它的返回值——这和常见的“从同步代码调用异步函数”场景的核心差异是必须拿到返回结果。

我曾尝试改成下面的写法，但始终触发TimeoutError，Future.result()无法正确返回结果：

loop = asyncio.get_running_loop()
obj = asyncio.run_coroutine_threadsafe(
    session.get(model_class, primary_key_value), loop
).result(timeout=5)

以下是我的最小测试代码，同样会报错：

import asyncio


async def main():
    """main是异步函数，通过asyncio.run正常启动"""
    print("BEGIN main")

    loop = asyncio.get_running_loop()
    timeout = 3
    # 创建协程
    coro = asyncio.sleep(1, result=3)

    # 将协程提交到指定事件循环
    future = asyncio.run_coroutine_threadsafe(coro, loop)

    # 等待结果，设置超时时间
    assert future.result(timeout) == 3


if __name__ == "__main__":
    asyncio.run(main())

核心问题分析 #

测试代码和实际代码的问题根源是：asyncio.run_coroutine_threadsafe的设计目标是向不同线程的事件循环提交协程，但你当前的代码是在同一个线程的事件循环里调用它——这会直接导致死锁：future.result()会阻塞当前线程，而事件循环被这个阻塞操作卡住，根本无法执行提交的协程，最终触发超时错误。

正确的兼容实现方式 #

方案1：拆分异步/同步函数+统一入口（推荐） #

这种方式逻辑清晰，避免同步异步混用的坑，同时覆盖所有场景：

import asyncio
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import Session
from sqlalchemy import select, Obj

async def _async_func(session: AsyncSession, data):
    statement = select(Obj).where(Obj.id == data['id'])
    # 执行异步查询
    result = await session.execute(statement)
    obj = result.scalar_one()
    # 修改obj的逻辑
    return obj

def _sync_func(session: Session, data):
    statement = select(Obj).where(Obj.id == data['id'])
    # 执行同步查询
    result = session.execute(statement)
    obj = result.scalar_one()
    # 修改obj的逻辑
    return obj

def func_for_async_and_sync(session, data):
    if isinstance(session, AsyncSession):
        try:
            # 检查当前是否处于异步上下文
            asyncio.get_running_loop()
            # 异步环境下直接返回协程，由调用方await
            return _async_func(session, data)
        except RuntimeError:
            # 同步环境下调用异步Session，新开事件循环执行
            return asyncio.run(_async_func(session, data))
    else:
        # 同步Session直接调用同步逻辑
        return _sync_func(session, data)

方案2：异步函数内兼容同步Session（仅适用于异步环境调用） #

如果通用函数只会在异步环境中被调用，仅需要兼容传入的同步Session，可以用线程池包装同步操作，避免阻塞事件循环：

import asyncio
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import Session
from sqlalchemy import select, Obj

async def func_for_async_and_sync(session, data):
    statement = select(Obj).where(Obj.id == data['id'])
    if isinstance(session, AsyncSession):
        result = await session.execute(statement)
    else:
        # 用线程池执行同步查询，避免阻塞事件循环
        result = await asyncio.to_thread(session.execute, statement)
    obj = result.scalar_one()
    # 修改obj的逻辑
    return obj

修复测试代码（仅作原理演示，不推荐业务使用） #

如果一定要用run_coroutine_threadsafe，必须在不同线程中运行目标事件循环：

import asyncio
import threading

def run_loop_in_thread(loop):
    asyncio.set_event_loop(loop)
    loop.run_forever()

async def main():
    print("BEGIN main")
    # 新建一个事件循环并在单独线程启动
    new_loop = asyncio.new_event_loop()
    loop_thread = threading.Thread(target=run_loop_in_thread, args=(new_loop,), daemon=True)
    loop_thread.start()

    timeout = 3
    coro = asyncio.sleep(1, result=3)
    # 向另一个线程的事件循环提交协程
    future = asyncio.run_coroutine_threadsafe(coro, new_loop)
    assert future.result(timeout) == 3
    # 停止另一个线程的事件循环并等待线程结束
    new_loop.call_soon_threadsafe(new_loop.stop)
    loop_thread.join()

if __name__ == "__main__":
    asyncio.run(main())

这种方式复杂度高且易出问题，不建议在业务代码中使用。

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