Async Fundamentals

FastAPI is built on an async foundation. Understanding async/await is not optional — using sync code incorrectly inside async routes is one of the most common production mistakes, causing request queuing and latency spikes under load.

Learning Focus

By the end of this lesson you can: explain the Python event loop model, decide when to use async def vs def, and safely call both async and sync libraries from FastAPI routes.

Sync vs Async — The Core Difference

In synchronous code, when your route calls a database, the entire thread waits for the response. Nothing else can run on that thread.

In async code, the event loop suspends the coroutine at each await point and handles other requests while waiting for I/O. One thread handles many concurrent requests.

In the synchronous model, Request 2 cannot start until Request 1 finishes — including the DB wait.

async def vs def in FastAPI

FastAPI supports both, but they behave differently:

	`async def`	`def`
Runs in	Event loop directly	Thread pool (via `run_in_executor`)
Blocks event loop on slow I/O	✅ Yes (if no `await`)	❌ No (thread pool)
Await async libraries	✅	❌ Syntax error
Overhead	Minimal	Thread pool overhead
Use when	Calling `await`-able libraries	Calling sync-only libraries

warning

Never call blocking I/O (sync DB drivers, requests, time.sleep, file reads) inside an async def route without putting it in a thread pool. Doing so blocks the event loop and serializes all requests.

When to Use `async def`

Use async def when you are calling async-native libraries:

app/routes/items.py
import httpx
from fastapi import FastAPI

app = FastAPI()

# ✅ Correct: async HTTP client, awaited properly
@app.get("/remote-data")
async def get_remote() -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.get("https://httpbin.org/json")
        return response.json()

# ✅ Correct: async SQLAlchemy session
from sqlalchemy.ext.asyncio import AsyncSession
from fastapi import Depends

async def get_db() -> AsyncSession: ...

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)) -> dict:
    result = await db.execute(...)  # non-blocking
    return result.scalar()

When to Use `def`

Use def when the library is sync-only and cannot be made async:

app/routes/legacy.py
import time
from fastapi import FastAPI

app = FastAPI()

# ✅ FastAPI runs this in a thread pool automatically
@app.get("/slow-sync")
def slow_sync_route() -> dict:
    time.sleep(1)  # Blocks, but only one thread, not the event loop
    return {"done": True}

# ❌ NEVER do this — blocks the entire event loop
@app.get("/broken-async")
async def broken_async() -> dict:
    time.sleep(1)  # This freezes ALL concurrent requests
    return {"done": True}

Running Sync Code from Async Routes

If you must call a blocking library from an async def route, use asyncio.to_thread (Python 3.9+):

app/routes/mixed.py
import asyncio
import time
from fastapi import FastAPI

app = FastAPI()

def blocking_operation(seconds: int) -> str:
    time.sleep(seconds)
    return f"Slept for {seconds}s"

@app.get("/safe-sync")
async def safe_sync() -> dict:
    # ✅ Runs in thread pool, event loop stays free
    result = await asyncio.to_thread(blocking_operation, 1)
    return {"result": result}

Making External HTTP Calls — Always Use httpx

Never use the requests library in async routes:

Library	Async?	Use in `async def`?
`requests`	❌ Sync	❌ Blocks event loop
`httpx`	✅ Async	✅ `await client.get(...)`
`aiohttp`	✅ Async	✅ Alternative

app/services/external.py
import httpx
from typing import Any

# ✅ Recommended pattern: reuse a client via lifespan
_client: httpx.AsyncClient | None = None

async def get_client() -> httpx.AsyncClient:
    if _client is None:
        raise RuntimeError("Client not initialized")
    return _client

async def fetch_json(url: str) -> Any:
    client = await get_client()
    response = await client.get(url, timeout=10.0)
    response.raise_for_status()
    return response.json()

Async Context Manager Pattern

For resources that need setup/teardown (DB connections, HTTP clients), use FastAPI's lifespan:

app/main.py
from contextlib import asynccontextmanager
import httpx
from fastapi import FastAPI

http_client: httpx.AsyncClient | None = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global http_client
    # Startup
    http_client = httpx.AsyncClient()
    yield
    # Shutdown
    await http_client.aclose()

app = FastAPI(lifespan=lifespan)

Common Pitfalls

Pitfall	Cause / Symptom	Fix
`requests.get()` in `async def`	Event loop blocked, latency spikes under load	Replace with `httpx.AsyncClient` and `await`
`time.sleep()` in `async def`	All concurrent requests queue behind it	Use `await asyncio.sleep()` or move to `def`
`sqlalchemy` sync session in `async def`	DB queries serialize all requests	Use `AsyncSession` with `asyncpg` driver
Forgetting `await` on a coroutine	Coroutine object returned instead of value	All `async def` calls need `await`
Using `async def` for CPU-bound work	Event loop is still blocked	Move CPU work to `ProcessPoolExecutor`

Hands-On Practice

app/async_demo.py
import asyncio
import httpx
from fastapi import FastAPI

app = FastAPI(title="Async Demo")

@app.get("/sleep")
async def async_sleep(seconds: float = 0.5) -> dict:
    """Non-blocking sleep — other requests run during this."""
    await asyncio.sleep(seconds)
    return {"slept": seconds}

@app.get("/fetch")
async def fetch_example() -> dict:
    """Fetch external data without blocking."""
    async with httpx.AsyncClient() as client:
        resp = await client.get("https://httpbin.org/get", timeout=5.0)
        data = resp.json()
    return {"origin": data.get("origin"), "status": resp.status_code}

@app.get("/concurrent")
async def concurrent_example() -> dict:
    """Run two async operations simultaneously."""
    async with httpx.AsyncClient() as client:
        task1 = client.get("https://httpbin.org/delay/1", timeout=5.0)
        task2 = client.get("https://httpbin.org/delay/1", timeout=5.0)
        # Both run concurrently — total ~1s, not 2s
        r1, r2 = await asyncio.gather(task1, task2)
    return {"r1_status": r1.status_code, "r2_status": r2.status_code}

run-async-demo.sh
uvicorn app.async_demo:app --reload --port 8002

# Test non-blocking sleep
curl http://localhost:8002/sleep?seconds=1

# Test concurrent fetch (should be ~1s not 2s)
time curl http://localhost:8002/concurrent

Sync vs Async — The Core Difference​

async def vs def in FastAPI​

When to Use async def​

When to Use def​

Running Sync Code from Async Routes​

Making External HTTP Calls — Always Use httpx​

Async Context Manager Pattern​

Common Pitfalls​

Hands-On Practice​

What's Next​