Python异步转同步装饰器失效，如何修复实现？

问题背景 #

在基于FastAPI、MongoDB和Beanie的异步应用中，需要将Beanie的异步函数包装为同步函数，但自行编写的两个装饰器均失效：

第一个装饰器及问题

代码：

def async_to_sync(async_func):
    def wrapper(*args, **kwargs):
        loop = asyncio.get_running_loop()
        asyncio.set_event_loop(loop)
        result = loop.run_until_complete(async_func(*args, **kwargs))
        loop.close()
        return result

    return wrapper

报错：

File "uvloop/loop.pyx", line 1511, in uvloop.loop.Loop.run_until_complete
  File "uvloop/loop.pyx", line 1504, in uvloop.loop.Loop.run_until_complete
  File "uvloop/loop.pyx", line 1377, in uvloop.loop.Loop.run_forever
  File "uvloop/loop.pyx", line 518, in uvloop.loop.Loop._run
RuntimeError: this event loop is already running.

问题原因：FastAPI本身运行在异步事件循环中，get_running_loop()获取的是已启动的循环，而run_until_complete()不能在运行中的循环内调用；同时手动close()循环会破坏FastAPI的原生循环，导致后续异步任务崩溃。

第二个装饰器及问题

代码：

def async_to_sync(async_func):
    def wrapper(*args, **kwargs):
        result = asyncio.ensure_future(async_func(*args, **kwargs))
        return result

    return wrapper

问题原因：ensure_future()仅将协程加入事件队列，不会等待其执行完成，直接返回处于pending状态的Task对象，无法拿到实际结果。

可行的装饰器实现 #

方案1：线程安全的跨循环执行（无需额外依赖）

通过判断当前是否有运行中的事件循环，分别处理：

import asyncio
from threading import Thread

def async_to_sync(async_func):
    def wrapper(*args, **kwargs):
        try:
            loop = asyncio.get_running_loop()
            if loop.is_running():
                # 已有运行中的循环，用线程安全方法提交协程并等待结果
                future = asyncio.run_coroutine_threadsafe(async_func(*args, **kwargs), loop)
                return future.result()
        except RuntimeError:
            # 无运行中循环，直接用asyncio.run启动执行
            return asyncio.run(async_func(*args, **kwargs))
        
        # 兜底逻辑
        return asyncio.run(async_func(*args, **kwargs))
    
    return wrapper

方案2：使用nest_asyncio允许循环嵌套（更简洁）

先安装依赖：

pip install nest_asyncio

装饰器代码：

import asyncio
import nest_asyncio

# 全局应用补丁，允许事件循环嵌套
nest_asyncio.apply()

def async_to_sync(async_func):
    def wrapper(*args, **kwargs):
        loop = asyncio.get_event_loop()
        return loop.run_until_complete(async_func(*args, **kwargs))
    
    return wrapper

注意事项 #

• 方案1不修改原有事件循环行为，适合对循环稳定性要求高的场景，线程切换对Beanie的IO操作性能影响极小。
• 方案2更简洁，但nest_asyncio会全局修改事件循环的嵌套规则，需确保应用中没有依赖原生循环非嵌套特性的代码。

内容的提问来源于stack exchange，提问作者hrmb-hector