Async & Performance — Cheat Sheet

async def vs def — How FastAPI Runs Your Endpoints

import asyncio, time
from fastapi import FastAPI

app = FastAPI()

# A) def + blocking → thread pool handles it
@app.get("/report-def")
def report_def():
    time.sleep(1)                   # blocking, but only THIS thread waits
    return {"ok": True}
# 10 concurrent requests → ~1s total (10 threads sleep in parallel)

# B) async def + await → event loop handles it
@app.get("/report-async")
async def report_async():
    await asyncio.sleep(1)          # yields; loop serves others meanwhile
    return {"ok": True}
# 10 concurrent → ~1s. 10,000 concurrent → still fine (no threads needed)

# C) async def + BLOCKING — the bug
@app.get("/report-broken")
async def report_broken():
    time.sleep(1)                   # ✗ blocks the event loop itself
    return {"ok": True}
# 10 concurrent requests → ~10s. Request #10 waited for all nine others.
# EVERY endpoint of the app is frozen during each sleep — /health too.

# Escape hatch when stuck with blocking code in an async path:
@app.get("/report-rescued")
async def report_rescued():
    await asyncio.to_thread(time.sleep, 1)   # push to a thread, keep loop free
    return {"ok": True}

# CPU-bound (ML inference, PDF parsing, crypto)?
# Threads don't help (GIL — Python track) and the loop must never do it:
# ProcessPoolExecutor via run_in_executor, or a task queue (next chapters).

Calling External APIs — httpx.AsyncClient Patterns

import asyncio, httpx
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http = httpx.AsyncClient(
        timeout=httpx.Timeout(5.0, connect=2.0),   # read 5s, connect 2s — ALWAYS explicit
        limits=httpx.Limits(max_connections=100, max_keepalive_connections=20),
    )
    yield
    await app.state.http.aclose()

app = FastAPI(lifespan=lifespan)

@app.get("/students/{sid}/dashboard")
async def dashboard(sid: int, request: Request):
    http = request.app.state.http          # ONE pool for the whole app

    # three independent upstreams — overlap them:
    profile_t = http.get(f"https://profile.internal/students/{sid}")
    offers_t  = http.get(f"https://offers.internal/students/{sid}/offers")
    events_t  = http.get(f"https://events.internal/upcoming")

    profile, offers, events = await asyncio.gather(
        profile_t, offers_t, events_t,
        return_exceptions=True,            # one failure ≠ whole page failure
    )

    def safe_json(r, fallback):
        return r.json() if isinstance(r, httpx.Response) and r.is_success else fallback

    return {
        "profile": safe_json(profile, {}),          # required-ish
        "offers":  safe_json(offers, []),           # degrade to empty
        "events":  safe_json(events, []),           # degrade to empty
    }
# Sequential: 300+300+300 = 900ms. gather: max(300,300,300) = 300ms.
# Client-per-request instead of shared: add a TCP+TLS handshake to every call.

Caching — Redis, TTLs & Invalidation

# pip install redis
import json
import redis.asyncio as aioredis
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.redis = aioredis.from_url("redis://localhost:6379/0",
                                        decode_responses=True)
    yield
    await app.state.redis.aclose()

app = FastAPI(lifespan=lifespan)

CACHE_VER = "v2"          # bump when the cached SHAPE changes (see below)

def drives_key(branch: str) -> str:
    return f"placement:{CACHE_VER}:drives:open:{branch}"     # structured, greppable

@app.get("/drives/open")
async def open_drives(branch: str, request: Request):
    r = request.app.state.redis
    key = drives_key(branch)

    if (cached := await r.get(key)) is not None:             # 1. try cache
        return json.loads(cached)                            #    sub-ms hit

    drives = await expensive_drives_query(branch)            # 2. miss → DB (200ms)
    await r.setex(key, 300, json.dumps(drives))              # 3. store, TTL 5 min
    return drives

@app.post("/drives", status_code=201)
async def create_drive(payload: dict, request: Request):
    drive = await insert_drive(payload)                      # write to DB first
    # 4. INVALIDATE — readers rebuild on next request:
    await request.app.state.redis.delete(drives_key(payload["branch"]))
    return drive

# TTL doctrine: every key gets one, even with explicit invalidation —
# TTL is the safety net for the delete you forgot. Hot + rarely changing
# → longer TTL; user-visible-fresh (seat counts) → 10-30s micro-TTL still
# absorbs 99% of reads at 1000 rps.