python-3.x 如何立即从asyncio.Task获取异常?

yhived7q  于 2023-06-25  发布在  Python
关注(0)|答案(1)|浏览(89)
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只在脚本停止时显示,这并不太方便。是否有一种方法可以在异常引发时立即显示它?

oxf4rvwz

oxf4rvwz1#

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)

相关问题