Async and Background Tasks

Introduction

FastAPI supports two kinds of endpoint functions: normal def and async def. Choosing correctly affects performance and, if you choose wrongly, can freeze your whole server. This lesson explains the idea with a measured example.

Waiting is the problem

Much API work is waiting: for a database, another API or a disk. While one request waits, the server could serve others. Async code lets a single thread switch between tasks whenever one is waiting. The mechanism is an event loop and the keywords async and await.

The rule of thumb

• Use async def when your code calls async libraries and uses await: async database drivers, async HTTP clients.
• Use plain def when you call normal blocking libraries: a synchronous database library, file processing, requests. FastAPI runs def endpoints in a thread pool, so they do not block the event loop.
• If you are unsure, use def. It is safe.

The big mistake: calling a blocking function, such as time.sleep(5) or a slow synchronous HTTP call, inside async def. It blocks the event loop, and every other request waits until it finishes. Use await asyncio.sleep(...) or an async library, or put the work in a def endpoint.

Concurrency with gather

If an endpoint needs several independent slow results, start them together.

import asyncio
import time


async def slow_lookup(n: int) -> int:
    await asyncio.sleep(0.2)      # pretend this is a network call
    return n * 2


@app.get("/gather")
async def gather():
    start = time.perf_counter()
    results = await asyncio.gather(slow_lookup(1), slow_lookup(2), slow_lookup(3))
    return {"results": results, "seconds": round(time.perf_counter() - start, 2)}

Awaited one after another, three 0.2-second calls would take about 0.6 seconds. With gather they overlap and take about 0.2 seconds. When I ran it, the results were [2, 4, 6] and the total was under 0.5 seconds.

Concurrency helps with waiting. It does not speed up heavy calculation: CPU-bound work needs separate processes or workers.

Background tasks

Sometimes you want to respond immediately and do something afterwards, such as writing a log or sending an email.

from fastapi import BackgroundTasks


def write_log(message: str):
    with open("log.txt", "a") as f:
        f.write(message + "\n")


@app.post("/notify")
def notify(background: BackgroundTasks, user: Annotated[str, Depends(current_user)]):
    background.add_task(write_log, f"notified {user}")
    return {"queued": True}

FastAPI sends the response, then runs write_log. In my run the endpoint returned {'queued': True} and the task then appended notified ada.

Limits of BackgroundTasks:

• It runs in the same process. If the server stops, queued work is lost.
• There are no retries, no progress tracking, and no scheduling.
• Long or critical jobs belong in a proper task queue such as Celery, RQ or a cloud queue, with a separate worker.

Resource management

• Reuse HTTP clients and database connections instead of creating new ones for each request.
• Set timeouts on outbound calls, so a slow service cannot hang your API.
• Handle errors in background tasks. Failures there are not shown to the client.

Try it yourself

Create two endpoints that each wait 1 second, one using asyncio.sleep in async def, and one using time.sleep in async def. Send several requests at once and compare the total time. Then change the second to plain def and measure again.

Summary

• Use async def with async libraries and plain def with blocking libraries, and never block the event loop.
• asyncio.gather overlaps independent waits, but does not speed up CPU-heavy work.
• BackgroundTasks suit small after-response jobs, while important work needs a real queue.