Async / Await — When and How¶

FastAPI is async-native. But what does that mean — and when should YOU use async def instead of def?

The problem async solves¶

When your endpoint does I/O — calls a database, hits another API, reads a file — it spends most of its time waiting, not computing.

Without async (Flask-style):

Request 1 → wait 500ms for DB → respond
Request 2 → BLOCKED until Request 1 finishes → wait 500ms for DB → respond
Total: 1000ms for 2 requests

With async:

Request 1 → wait for DB...
Request 2  → wait for DB...   (started while Request 1 was waiting)
Both responses arrive in ~500ms

The CPU isn't doing anything during the wait. Async lets the same process handle other requests while waiting.

Sync vs async endpoint¶

from fastapi import FastAPI
import time, asyncio

app = FastAPI()

# Sync — blocks the worker for 2 seconds
@app.get("/sync")
def sync_endpoint():
    time.sleep(2)
    return {"message": "synchronous"}

# Async — frees the worker during the 2-second wait
@app.get("/async")
async def async_endpoint():
    await asyncio.sleep(2)
    return {"message": "asynchronous"}

Hit /sync from 10 browser tabs at once → they finish one after another (20s total). Hit /async from 10 tabs → they all finish around 2s.

Important: asyncio.sleep() is async-friendly. time.sleep() is not — it blocks the whole event loop. Same applies to requests.get() (blocking) vs httpx.AsyncClient (async).

Rule of thumb — when to use `async def`¶

You're doing	Use
Pure Python computation (math, sorting, parsing)	`def`
Database query with an async driver	`async def`
HTTP call with `httpx.AsyncClient`	`async def`
File read with `aiofiles`	`async def`
Calling a sync library (regular `requests`, regular SQLAlchemy, blocking SDK)	`def`
Mixed	If your dependencies are sync → use `def`

Don't mark an endpoint async def if you call only sync code inside. FastAPI will run sync def endpoints in a threadpool — works fine for blocking work.

A real-world async example with `httpx`¶

from fastapi import FastAPI
import httpx

app = FastAPI()

@app.get("/user/{username}")
async def github_user(username: str):
    async with httpx.AsyncClient() as client:
        r = await client.get(f"https://api.github.com/users/{username}")
    if r.status_code != 200:
        return {"error": "not found"}
    data = r.json()
    return {
        "name": data.get("name"),
        "bio": data.get("bio"),
        "followers": data.get("followers"),
    }

async with and await mean the request handler can do other work while the network call is pending.

Concurrent calls with `asyncio.gather`¶

Need to call multiple APIs at once and combine?

import asyncio
import httpx

@app.get("/dashboard/{username}")
async def dashboard(username: str):
    async with httpx.AsyncClient() as client:
        user_task = client.get(f"https://api.github.com/users/{username}")
        repos_task = client.get(f"https://api.github.com/users/{username}/repos")
        gists_task = client.get(f"https://api.github.com/users/{username}/gists")
        user, repos, gists = await asyncio.gather(user_task, repos_task, gists_task)

    return {
        "user": user.json().get("name"),
        "repo_count": len(repos.json()),
        "gist_count": len(gists.json()),
    }

Three HTTP calls happen in parallel. Total time ≈ slowest single call, not sum of all three.

Background tasks¶

For "fire and forget" — send email, log analytics — use BackgroundTasks:

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_welcome_email(email: str):
    # imagine this calls an email service — slow
    print(f"📧 sent welcome to {email}")

@app.post("/signup")
def signup(email: str, background_tasks: BackgroundTasks):
    # ... save user to DB ...
    background_tasks.add_task(send_welcome_email, email)
    return {"status": "ok"}        # returns immediately

The response is sent first. The email task runs after the response is delivered. The user doesn't wait.

For heavier or persistent jobs, use a real queue: Celery, RQ, or Arq (async). BackgroundTasks is in-process and goes away if your server restarts.

Async generators — streaming¶

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio

app = FastAPI()

async def slow_stream():
    for i in range(5):
        await asyncio.sleep(1)
        yield f"chunk {i}\n"

@app.get("/stream")
async def stream():
    return StreamingResponse(slow_stream(), media_type="text/plain")

Application lifespan — startup / shutdown hooks¶

For things that should run when your app starts or stops:

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # --- startup ---
    print("🔌 connecting to DB")
    app.state.db = await connect_db()           # imagine this exists
    yield
    # --- shutdown ---
    print("🔌 closing DB")
    await app.state.db.close()

app = FastAPI(lifespan=lifespan)

@app.get("/health")
async def health():
    return {"db_connected": app.state.db.is_alive()}

How a request flows in async mode¶

1. uvicorn (event loop) receives request
        ↓
2. FastAPI finds matching route
        ↓
3. Parses path/query/body, validates
        ↓
4. Schedules your async function on the event loop
        ↓
5. Function runs. At `await ...`, control returns to event loop.
        ↓
6. Event loop handles OTHER incoming requests while waiting.
        ↓
7. When the awaited thing finishes, your function resumes.
        ↓
8. Function returns → response sent.

Sync def endpoints follow almost the same flow, except step 5 runs in a worker thread instead of on the event loop.

Common pitfalls¶

❗ Calling sync code inside async — time.sleep(2) inside async def blocks the whole event loop. Use await asyncio.sleep(2) instead.
❗ Calling sync libraries inside async def — requests.get(...) inside async def blocks the event loop. Either: (a) switch to httpx.AsyncClient, or (b) just use plain def (FastAPI runs it in a threadpool).
❗ async def for pure CPU work — adds overhead with no benefit. CPU-bound work belongs in a process pool, not async.
❗ Forgetting await — client.get(url) (no await) returns a coroutine, not a response. You'll get cryptic errors.
❗ Mixing sync DB drivers with async def — every query blocks. Either use an async driver (asyncpg, motor, aiosqlite) or just use sync.

What's next¶

Practice¶

What does this print?

Expected: done

import asyncio
async def task():
    return "done"
print(asyncio.run(task()))

Don't call a blocking function (time.sleep) inside an async handler

Expected: True

import asyncio
async def handler():
    import time
    time.sleep(0.01)            # bug: blocks the event loop; use await asyncio.sleep instead
    return "done"
result = asyncio.run(handler())
print(result == "done")

Quiz — Quick check¶

What you remember

Q1. When should a FastAPI endpoint be async def?

When it awaits I/O (database, HTTP calls, file operations) — async lets other requests run during the wait
Always
Never
Only for streaming

Why: Async wins for I/O-bound work (most APIs). For CPU-bound work, async doesn't help and may hurt — FastAPI runs sync handlers in a thread pool, async handlers in the event loop.

Q2. What's wrong with calling time.sleep(1) in an async handler?

Blocks the entire event loop — no other requests can be served for that second
No effect
Slower than await asyncio.sleep(1)
Crashes the server

Why: time.sleep is synchronous. Inside async def, it blocks the event loop. Use await asyncio.sleep(1) instead — gives control back so other tasks can run.

Q3. Can I use a synchronous database driver in an async handler?

Yes but it blocks the event loop — prefer an async driver (asyncpg, motor) when possible
No, never
Yes, no downsides
Only with workers

Why: A sync driver inside async def blocks the loop while the query runs — losing the concurrency benefit. Async drivers cooperate properly. If you must use sync, wrap with run_in_executor.

Common doubts¶

How do I tell if I should use sync or async for a handler?

Rule of thumb: if your handler awaits anything (DB, HTTP, file I/O), use async def. If it's all CPU (calculations, ML inference), use plain def — FastAPI runs it in a thread pool. Mixing is fine; pick per-handler.

Why is my async endpoint slow?

Common cause: blocking calls inside async def. Look for time.sleep, sync HTTP libs (requests), sync DB calls. Replace with async equivalents: asyncio.sleep, httpx, asyncpg. Or wrap blocking calls in loop.run_in_executor().

Can I call sync code from async code?

Yes — direct calls work but block the loop. For long-running sync code: await asyncio.to_thread(my_sync_function, arg1, arg2). This runs the sync function in a thread pool while the event loop continues.

→ Dependency Injection