Python中syncify/asyncify语法糖通用方案及PEP化探讨

需求背景 #

需实现两类简洁互调场景，完全隐藏async/await底层实现细节：

• 从同步上下文直接调用异步函数
• 从异步上下文直接调用同步函数

现有方案局限性 #

已尝试的四类方案均无法作为通用语法糖，需开发者手动判断运行上下文：

• 方案A（2014年syncify项目）：Python 3.10中触发SyntaxError，版本兼容性差
• 方案B（某Gist实现）：执行scite_entry=force_sync(doi_obj.asScite)()时抛出RuntimeError: 事件循环已运行，无法处理已有事件循环的场景
• 方案C（asgiref.AsyncToSync）：抛出RuntimeError: 不可在异步事件循环线程中使用，仅支持同步→异步单向调用，无法在异步上下文复用
• 方案D（asyncer.syncify）：执行scite_entry=syncify(doi_obj.asScite)()时抛出RuntimeError: 仅可在AnyIO工作线程运行，依赖特定异步框架，通用性不足

通用实现方案 #

核心思路 #

通过自动检测当前运行上下文（是否存在活跃事件循环），封装适配逻辑：

1. 同步上下文调用异步函数：自动创建/复用事件循环执行异步任务
2. 异步上下文调用同步函数：自动通过线程池执行同步任务，避免阻塞事件循环

代码实现 #

import asyncio
from concurrent.futures import ThreadPoolExecutor
from functools import wraps

# 全局线程池，复用资源
_sync_executor = ThreadPoolExecutor()

def auto_adapt(func):
    """自动适配同步/异步函数的调用上下文"""
    if asyncio.iscoroutinefunction(func):
        # 适配异步函数：支持在同步上下文直接调用
        @wraps(func)
        def sync_wrapper(*args, **kwargs):
            try:
                # 尝试获取已有事件循环
                loop = asyncio.get_running_loop()
                if loop.is_running():
                    # 事件循环已运行，通过run_until_complete获取结果
                    return loop.run_until_complete(func(*args, **kwargs))
            except RuntimeError:
                # 无活跃事件循环，创建新事件循环执行
                return asyncio.run(func(*args, **kwargs))
            # 若在异步上下文，直接返回协程（由调用方await）
            return func(*args, **kwargs)
        return sync_wrapper
    else:
        # 适配同步函数：支持在异步上下文直接await
        @wraps(func)
        async def async_wrapper(*args, **kwargs):
            loop = asyncio.get_running_loop()
            return await loop.run_in_executor(_sync_executor, func, *args, **kwargs)
        return async_wrapper

使用示例 #

1. 同步上下文调用异步函数

async def async_task():
    await asyncio.sleep(1)
    return "异步任务完成"

# 被装饰后，同步代码可直接调用
adapted_async = auto_adapt(async_task)
print(adapted_async())  # 输出：异步任务完成

2. 异步上下文调用同步函数

def sync_task():
    import time
    time.sleep(1)
    return "同步任务完成"

# 被装饰后，异步代码可直接await
adapted_sync = auto_adapt(sync_task)

async def main():
    result = await adapted_sync()
    print(result)  # 输出：同步任务完成

asyncio.run(main())

PEP纳入可行性分析 #

该方案完全贴合PEP-0492的简洁性目标：

• 无需开发者手动处理事件循环、线程池等底层细节
• 自动适配运行上下文，降低心智负担
• 实现逻辑轻量，不依赖第三方框架，可作为标准库asyncio模块的辅助工具或语法糖提案

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