How to instantly get exceptions from asyncio.Task?

Question

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)

When I run this code, the ValueError is only displayed when the script is stopped, which isn't very convenient. Is there a way to display the exception as soon as it is raised?

Answer 1

To do something as soon as a task done (successfully or not) is just what asyncio.Task.add_done_callback for:

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)