Python 3.12中能否实现支持同步/异步调用的方法?
在Python 3.12中实现同步/异步兼容的方法
可以通过编写一个装饰器,让异步函数同时支持直接同步调用和异步await调用,无需调用者手动使用asyncio.run()。核心思路是自动检测当前是否处于异步上下文:如果是,返回协程对象供调用者await;如果不是,则自动启动事件循环执行协程并返回结果。
实现方案
1. 编写同步/异步兼容装饰器
import asyncio from functools import wraps def sync_async(func): @wraps(func) def wrapper(*args, **kwargs): try: # 检查当前是否有正在运行的事件循环 loop = asyncio.get_running_loop() if loop.is_running(): # 处于异步上下文,返回协程对象 return func(*args, **kwargs) except RuntimeError: # 无运行中的事件循环,同步执行协程 return asyncio.run(func(*args, **kwargs)) # 处理事件循环存在但未运行的边缘情况 return asyncio.run(func(*args, **kwargs)) return wrapper
2. 用装饰器修饰目标函数
@sync_async async def maybe_async(something_that_makes_it_async=None): if something_that_makes_it_async: # 模拟异步操作(比如HTTP请求、IO任务) await asyncio.sleep(0.5) return f"异步结果来自 {something_that_makes_it_async}" else: # 同步逻辑处理 return "同步结果"
调用示例
同步调用(无需await)
res = maybe_async() print(res) # 输出:同步结果
异步调用(需要await)
async def main(): res = await maybe_async("some_http_interface") print(res) # 输出:异步结果来自 some_http_interface asyncio.run(main())
关键说明
- 这个方案在Python 3.7+都能正常工作,Python 3.12没有引入专门的语法糖,但该实现完全兼容3.12环境。
- 装饰器通过
asyncio.get_running_loop()检测上下文:抛出RuntimeError时说明处于同步环境,自动用asyncio.run()执行协程;如果检测到运行中的事件循环,则直接返回协程对象,避免在异步上下文里嵌套启动事件循环导致的错误。 - 注意:在异步上下文中调用时必须使用
await,否则会得到一个未执行的协程对象(Python会发出警告)。
内容的提问来源于stack exchange,提问作者steba92
相关产品推荐
相关产品推荐

