如何编写可在同步/异步上下文通用的asyncio调用包装函数？

问题概述 #

我使用dbus_next（基于asyncio的DBus库）访问DBus API，但项目中同时存在同步/线程代码，因此封装了同步调用函数。该函数在同步上下文可正常运行，但在异步上下文调用时会抛出RuntimeError: This event loop is already running错误。由于调用链路层级复杂，维护异步/同步两套代码不现实，需要实现一个可同时在两种上下文调用的包装函数。

原封装代码：

intfc = "..."
path = "..."

async def dbus_call_async():
    bus = await dbus_next.aio.MessageBus(bus_type=dbus_next.BusType.SYSTEM).connect()
    introspection = await bus.introspect(intfc, path)
    obj = bus.get_proxy_object(intfc, path, introspection)
    return await obj.call_my_api()

def dbus_call():
    return asyncio.get_event_loop().run_until_complete(dbus_call_async())

最小复现示例及报错：

import asyncio
import sys

async def async_call():
    return "Hello, world"

def _run_asyncio(coro):
    loop = asyncio.get_event_loop()
    result = asyncio.get_event_loop().run_until_complete(coro)
    return result

def wrapped_call():
    return _run_asyncio(async_call())

async def async_main():
    print(wrapped_call())   

def sync_main():
    print(wrapped_call())

if len(sys.argv) > 1:
    print("Calling in asyncio context")
    asyncio.get_event_loop().run_until_complete(async_main())
else:
    print("Calling in sync context")
    sync_main()

异步上下文调用时的报错：

Calling in asyncio context
Traceback (most recent call last):
  File "/home/tkcook/test2.py", line 23, in <module>
    asyncio.get_event_loop().run_until_complete(async_main())
  File "/usr/lib/python3.10/asyncio/base_events.py", line 649, in run_until_complete
    return future.result()
  File "/home/tkcook/test2.py", line 16, in async_main
    print(wrapped_call())   
  File "/home/tkcook/test2.py", line 13, in wrapped_call
    return _run_asyncio(async_call())
  File "/home/tkcook/test2.py", line 9, in _run_asyncio
    result = asyncio.get_event_loop().run_until_complete(coro)
  File "/usr/lib/python3.10/asyncio/base_events.py", line 625, in run_until_complete
    self._check_running()
  File "/usr/lib/python3.10/asyncio/base_events.py", line 584, in _check_running
    raise RuntimeError('This event loop is already running')
RuntimeError: This event loop is already running

解决方案 #

核心思路是自动判断当前执行上下文：如果处于同步上下文（无运行中的事件循环），则阻塞执行协程并返回结果；如果处于异步上下文（事件循环已运行），则返回协程对象，由调用者通过await获取结果。这种方式仅需维护一套核心异步逻辑，无需修改所有调用链路（仅异步上下文的调用处需添加await）。

通用适配函数 #

首先实现一个兼容Python 3.6的协程执行适配函数：

import asyncio

def run_coroutine(coro):
    # 检查当前是否处于异步上下文（Python 3.6兼容）
    current_task = asyncio.Task.current_task()
    if current_task is not None:
        # 异步上下文：返回协程，由调用者await
        return coro
    else:
        # 同步上下文：阻塞执行协程并返回结果
        loop = asyncio.get_event_loop()
        return loop.run_until_complete(coro)

适配DBus调用的包装函数 #

修改原DBus调用包装：

intfc = "..."
path = "..."

async def dbus_call_async():
    bus = await dbus_next.aio.MessageBus(bus_type=dbus_next.BusType.SYSTEM).connect()
    introspection = await bus.introspect(intfc, path)
    obj = bus.get_proxy_object(intfc, path, introspection)
    return await obj.call_my_api()

def dbus_call():
    return run_coroutine(dbus_call_async())

修改后的最小复现示例 #

import asyncio
import sys

async def async_call():
    return "Hello, world"

def run_coroutine(coro):
    current_task = asyncio.Task.current_task()
    if current_task is not None:
        return coro
    else:
        loop = asyncio.get_event_loop()
        return loop.run_until_complete(coro)

def wrapped_call():
    return run_coroutine(async_call())

async def async_main():
    # 异步上下文调用需添加await
    print(await wrapped_call())   

def sync_main():
    # 同步上下文直接调用
    print(wrapped_call())

if len(sys.argv) > 1:
    print("Calling in asyncio context")
    asyncio.get_event_loop().run_until_complete(async_main())
else:
    print("Calling in sync context")
    sync_main()

关键说明 #

1. Python 3.6兼容性：使用asyncio.Task.current_task()判断当前是否处于异步上下文（Python 3.7+可使用asyncio.get_running_loop()，但3.6不支持）。
2. 上下文适配逻辑：
   • 同步上下文：直接调用run_until_complete阻塞执行协程，返回结果，保持原有同步调用逻辑不变。
   • 异步上下文：返回协程对象，调用者只需添加await即可获取结果，无需重构整个调用链路。
3. 无需维护两套代码：核心逻辑仅需维护异步版本dbus_call_async，包装函数自动适配两种上下文。

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