Python3.x 如何立即从asyncio.Task获取异常

发布于06月17日

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.Task.add_done_callback的意思:

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())

        # here:
        self._foo_func_task.add_done_callback(
            lambda task: print(f"Exception: {type(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)