开启return_exceptions后如何检测asyncio.gather返回的异常并设置默认值

当return_exceptions=True时，asyncio.gather的返回列表中，执行失败的任务结果不会触发异常抛出，而是直接把对应异常的实例对象作为列表元素存储。要判断结果里是否存在异常，核心就是遍历结果列表，检查每一个元素是不是异常类的实例。

可直接复用的实现代码 #

import asyncio

# 自定义预设默认值
DEFAULT_RESPONSE = {"status": "failed", "data": None}

async def run_tasks():
    task_list = [task1(), task2(), task3()]
    # gather返回的是awaitable对象，必须await才能拿到聚合后的实际结果
    gather_awaitable = asyncio.gather(*task_list, return_exceptions=True)
    res = await gather_awaitable

    # 逐元素检查是否为异常实例
    has_exception = any(isinstance(item, BaseException) for item in res)
    response = DEFAULT_RESPONSE if has_exception else res
    return response

关键注意事项 #

• 必须对gather返回的对象执行await操作，不能直接对未完成的协程对象做异常检查，否则会出现完全错误的判断结果
• 不要用item is Error这类写法做判断：列表中存储的是异常的实例，不是异常类本身，直接做相等匹配永远无法命中
• 如果需要区分不同类型的异常（比如仅将业务自定义异常判定为可替换默认值的场景，忽略任务取消类的系统级异常），可以缩小isinstance的判断范围，比如仅检查isinstance(item, YourBusinessCustomException)

实际运行效果参考：如果task2执行时抛出ValueError("入参格式不合法")，最终拿到的res列表结构为：
[task1正常返回值, ValueError("入参格式不合法"), task3正常返回值]
上述检查逻辑会直接识别出列表中的异常实例，触发返回默认值的分支，完全匹配提到的伪代码逻辑。

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