Python库同步异步兼容：混合调用场景下的问题与优化思路

针对你遇到的事件循环误判、同步/异步调用冲突问题，这里提供一个无需拆分函数的透明兼容方案，核心是通过调用栈检测区分同步/异步调用场景，结合线程执行解决已有事件循环时的同步阻塞需求。

核心思路 #

1. 识别调用方类型：通过检查调用栈帧，判断直接调用方是否为协程函数，而非仅检测当前是否处于事件循环中。
2. 已有事件循环时的同步执行：使用asyncio.to_thread()（Python 3.9+）将异步函数委托到线程池执行，避免阻塞主事件循环，同时同步返回结果。

实现代码 #

import asyncio
import inspect
from functools import wraps

def sync_async_compatible(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        # 获取直接调用方的代码对象
        caller_frame = inspect.currentframe().f_back
        caller_code = caller_frame.f_code
        # 判断调用方是否为协程函数
        is_caller_coroutine = inspect.iscoroutinefunction(caller_code)

        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            # 无运行中的事件循环，直接用asyncio.run执行并返回结果
            return asyncio.run(func(*args, **kwargs))
        
        if is_caller_coroutine:
            # 调用方是协程，返回协程对象让调用方自行await
            return func(*args, **kwargs)
        else:
            # 同步调用方但处于事件循环中，用线程执行异步函数并阻塞获取结果
            return loop.run_until_complete(asyncio.to_thread(func, *args, **kwargs))
    
    # 保留原函数的协程标识，确保inspect等工具能正确识别
    wrapper._is_coroutine = asyncio.coroutines._is_coroutine
    return wrapper

使用示例 #

# 你的异步API函数
@sync_async_compatible
async def api_call():
    # 模拟远程API调用逻辑
    await asyncio.sleep(0.5)
    return {"status": "ok", "data": "sample_result"}

# 场景1：普通同步脚本调用
print(api_call())  # 输出: {'status': 'ok', 'data': 'sample_result'}

# 场景2：异步函数中调用
async def async_workflow():
    result = await api_call()
    print(result)  # 输出: {'status': 'ok', 'data': 'sample_result'}

asyncio.run(async_workflow())

# 场景3：异步入口调用同步遗留代码，再调用库函数
def legacy_sync_code():
    # 这里是同步代码，调用库函数期望直接拿到结果
    return api_call()

async def async_entry_with_legacy():
    result = legacy_sync_code()
    print(result)  # 输出: {'status': 'ok', 'data': 'sample_result'}

asyncio.run(async_entry_with_legacy())

注意事项 #

• Python版本兼容：如果需要支持Python 3.8及以下，可用loop.run_in_executor(None, func, *args, **kwargs)替代asyncio.to_thread()，但需确保你的异步函数是线程安全的。
• 性能考量：调用栈检测会带来微小的性能开销，若你的库存在超高频调用场景，可考虑缓存调用方类型或优化检测逻辑。
• 线程安全：确保异步函数中的远程API调用、资源操作是线程安全的，避免多线程环境下的竞态问题。

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