如何立即捕获asyncio.Task抛出的异常？

import asyncio

from fastapi import FastAPI
import uvicorn


class FooClass:
    def __init__(self):
        self._foo_func_task: asyncio.Task = None

    async def start_foo_func(self):
        self._foo_func_task = asyncio.create_task(self.foo_func())

    async def foo_func(self):
        raise ValueError


app = FastAPI()


@app.on_event('startup')
async def startup_event():
    app.state.foo = FooClass()
    await app.state.foo.start_foo_func()


if __name__ == '__main__':
    uvicorn.run(app)

运行上述代码时，ValueError仅在脚本停止时才会显示，这并不便捷。请问是否有办法在异常抛出时立即显示它？

问题根源是asyncio.create_task创建的异步任务，若未被await或主动处理异常，异常只会在任务对象被垃圾回收时才触发输出。要实现异常抛出时立即显示，可通过以下两种方式处理：

方式一：给任务添加完成回调 #

修改start_foo_func方法，为创建的任务添加回调函数，在回调中检查并输出异常：

import asyncio
import traceback

from fastapi import FastAPI
import uvicorn


class FooClass:
    def __init__(self):
        self._foo_func_task: asyncio.Task = None

    def _handle_task_exception(self, task: asyncio.Task):
        try:
            task.result()
        except Exception:
            print("任务抛出异常：")
            traceback.print_exc()

    async def start_foo_func(self):
        self._foo_func_task = asyncio.create_task(self.foo_func())
        self._foo_func_task.add_done_callback(self._handle_task_exception)

    async def foo_func(self):
        raise ValueError("测试异常")


app = FastAPI()


@app.on_event('startup')
async def startup_event():
    app.state.foo = FooClass()
    await app.state.foo.start_foo_func()


if __name__ == '__main__':
    uvicorn.run(app)

方式二：在任务内部捕获并处理异常 #

直接在foo_func中捕获异常并输出，异常抛出时会立即被处理：

import asyncio
import traceback

from fastapi import FastAPI
import uvicorn


class FooClass:
    def __init__(self):
        self._foo_func_task: asyncio.Task = None

    async def start_foo_func(self):
        self._foo_func_task = asyncio.create_task(self.foo_func())

    async def foo_func(self):
        try:
            raise ValueError("测试异常")
        except Exception:
            print("任务抛出异常：")
            traceback.print_exc()


app = FastAPI()


@app.on_event('startup')
async def startup_event():
    app.state.foo = FooClass()
    await app.state.foo.start_foo_func()


if __name__ == '__main__':
    uvicorn.run(app)

方式一更适合统一处理多个任务的异常场景，方式二则更直观，适合单个任务的异常处理，两种方案都能实现异常即时输出的需求。

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