Keep Track of Background Tasks

A signup handler saves the new user, then sends a welcome email in the background so the response doesn't have to wait for it:

async def handle_signup(email: str) -> str:
    await save_user(email)
    asyncio.create_task(send_welcome_email(email))  # don't wait for it
    return "ok"

Starting a task and never looking at it again is called fire and forget. This code has two bugs. Python can delete the task before it finishes, and if it fails, nobody finds out.

The event loop holds tasks weakly

The event loop keeps only a weak reference to each task. A weak reference lets Python find an object without keeping it alive. If nothing else refers to the task, the garbage collector can delete it before it finishes. The Python documentation for create_task warns about exactly this.

You can watch it happen with a task that waits on a future nothing else refers to:

import asyncio
import gc


async def wait_forever() -> None:
    await asyncio.get_running_loop().create_future()


async def main() -> None:
    asyncio.create_task(wait_forever())  # no reference kept
    await asyncio.sleep(0)
    gc.collect()
    await asyncio.sleep(0)


asyncio.run(main())  # logs "Task was destroyed but it is pending!"

Real tasks are usually also reachable through whatever they wait on, such as a socket. So the bug is rare, and a test rarely catches it. Keep a reference anyway.

Keep each task in a set

The fix the documentation recommends is a set of tasks, and a done callback that removes each task when it finishes:

import asyncio
from collections.abc import Coroutine
from typing import Any

background_tasks: set[asyncio.Task[None]] = set()


def spawn(coro: Coroutine[Any, Any, None]) -> asyncio.Task[None]:
    task = asyncio.create_task(coro)
    background_tasks.add(task)
    task.add_done_callback(background_tasks.discard)
    return task

The set holds a strong reference, the ordinary kind that keeps the task alive. task.add_done_callback(fn) asks the event loop to call fn(task) after the task finishes. Here fn is background_tasks.discard, so the set holds only unfinished tasks. The callback runs on a later turn of the loop, so for a moment a finished task is still in the set.

Someone has to look at a failed task

When a task raises, it stores the exception. Nothing else happens until someone awaits the task or calls task.exception(). If nobody does, the error appears only when Python deletes the task object, as a log line:

Task exception was never retrieved

By then your program can't handle it. The set fixes the first bug but not this one, because the done callback removes the task without looking at it.

So background work needs an owner, an object that keeps the tasks. At a point you choose, the owner waits for them and checks how each one ended. Two tools help:

await asyncio.wait(tasks) waits until every task has finished, and raises none of their exceptions. It raises ValueError if tasks is empty.
task.exception() returns the task's exception, or None if it succeeded. A task is cancelled when something asked it to stop before it finished. On a cancelled task exception() raises CancelledError, so check task.cancelled() first.

import logging

logger = logging.getLogger(__name__)


async def drain(tasks: list[asyncio.Task[None]]) -> None:
    if not tasks:
        return
    await asyncio.wait(tasks)
    for task in tasks:
        if not task.cancelled() and (error := task.exception()) is not None:
            logger.error("background task failed", exc_info=error)

Decide what shutdown does

Before you start background work, decide what happens when the program stops. Either wait for the tasks to finish, or cancel them. In both cases, look at every result. When the tasks belong to one block of code, asyncio.TaskGroup does all of this for you. The next article covers it.

Check your understanding A task started with spawn raises ValueError, and the done callback removes it from the set. What happens to the error? Show answer

Nothing handles it. The callback only drops the reference. Nobody awaited the task or called exception(), so asyncio logs "Task exception was never retrieved" when Python deletes the task. To handle the error, give the tasks an owner that waits for them and calls exception() on each one.

See the note on keeping references in the create_task documentation.