Asynchronous Programming
Harness the full potential of FastAPI's asynchronous capabilities to build high-performance applications.
Content
Async and Await
Versions:
Watch & Learn
AI-discovered learning video
Sign in to watch the learning video for this topic.
Async and Await — The two little words that make FastAPI actually feel fast
"If synchronous code is a single-lane country road, async/await turns it into a multi-lane highway with toll booths — you still need rules, or everyone crashes into the snack shack."
You're continuing from Understanding Asynchronous I/O, so I won't re-declare what an event loop is like from scratch. Instead: think of this as the practical, slightly caffeinated workshop where you learn how to write endpoints that actually behave when they talk to databases, call external APIs, or do anything that waits.
What are async and await, in plain (and slightly dramatic) English?
async def: defines a coroutine function — a function that can pause and let other work run while it's waiting. It's the polite coworker that says "I'll wait here, you go ahead".await: used insideasync defto pause until an awaitable (another coroutine, a Task, or a Future) finishes.- Coroutine vs Task: a coroutine is the plan; a Task is the event loop scheduling that plan to actually run concurrently.
If
defis a phone call,async defis a Zoom meeting where you can multitask while someone else is talking — you mute/unmute withawait.
FastAPI endpoints: async def vs def (short version)
- Use
async defwhen your endpoint mostly does I/O-bound work: DB queries (via an async driver), HTTP calls, file/network I/O. - Use
def(sync) when you're calling CPU-bound libraries or blocking code you cannot change — FastAPI will spin that into a threadpool for you, but beware threadpool exhaustion.
Quick comparison table
| Endpoint type | When to use | Pros | Cons |
|---|---|---|---|
async def |
I/O-bound, async DB/HTTP | High concurrency, low latency under load | Needs async libs; blocking code kills performance |
def |
CPU-bound or blocking third-party libs | Easy to write, works with existing sync libs | Threadpool limits, less concurrency per worker |
Minimal examples (read: real code you can copy)
Async endpoint doing an external HTTP call with httpx (async):
from fastapi import FastAPI
import httpx
app = FastAPI()
@app.get('/external')
async def call_external():
async with httpx.AsyncClient(timeout=5) as client:
r = await client.get('https://httpbin.org/delay/1')
return r.json()
Sync endpoint accidentally blocking the event loop (DON'T do this):
import time
@app.get('/bad')
def blocking():
time.sleep(3) # blocks the worker's thread — if many requests arrive, threads fill up
return {"status": "done"}
CPU-bound work offloaded to a threadpool (safe option):
import asyncio
from concurrent.futures import ThreadPoolExecutor
executor = ThreadPoolExecutor()
async def compute_heavy(data):
loop = asyncio.get_running_loop()
return await loop.run_in_executor(executor, heavy_sync_function, data)
Real-world patterns (and when to use them)
- Concurrent external requests: use
asyncio.gatherto fan-out calls.
async def get_many(urls):
async with httpx.AsyncClient() as client:
tasks = [client.get(u) for u in urls]
responses = await asyncio.gather(*tasks)
return responses
Throttle concurrency:
asyncio.Semaphoreto prevent overwhelming the remote API or your DB connection pool.Timeouts & cancellation: use
asyncio.wait_forto bound your awaitable so hung calls don't hang your handler forever.
try:
data = await asyncio.wait_for(client.get(url), timeout=2)
except asyncio.TimeoutError:
# return a 504-ish response or fallback
pass
- Background & heavy jobs: for non-critical long-running tasks use FastAPI's BackgroundTasks or an external worker (Celery, RQ, etc.). Don't do huge CPU work inside the request handler.
Testing async endpoints — remember what we learned in "Using Test Clients"
Testing async code is only a little more magical:
- Use
httpx.AsyncClient(oranyio-backed fixtures) withpytest.mark.asyncioorpytest'sasyncio/anyiosupport. - In CI, ensure your test runner supports async (your pipelines previously set up for CI should install the right pytest plugins).
Example pytest async test:
import pytest
from httpx import AsyncClient
from myapp import app
@pytest.mark.asyncio
async def test_external_endpoint():
async with AsyncClient(app=app, base_url='http://test') as ac:
r = await ac.get('/external')
assert r.status_code == 200
If you used sync endpoints, TestClient from starlette still works synchronously — but prefer async clients to test async behavior properly (e.g., timeouts, concurrency).
Common pitfalls (read carefully)
- Blocking calls inside
async def— even one call to a blocking library (calls totime.sleep, blocking DB drivers, synchronous HTTP clients) will stall the event loop. Use async drivers orrun_in_executor. - Threadpool exhaustion — if lots of sync endpoints block threads, the app can stall. Monitor threadpool utilization.
- Mixing sync DB drivers: if your ORM only has sync drivers, either run DB access in a threadpool or migrate to an async-capable driver/ORM.
- Not limiting concurrency: launching hundreds of simultaneous external requests without limits can crash your client or the upstream service.
Best practices checklist (TL;DR)
- Prefer async/await for I/O-bound operations. Use async libraries end-to-end where possible (async DB drivers, httpx AsyncClient).
- Keep handlers small; delegate heavy work to background tasks or workers.
- Use timeouts and graceful cancellation (
asyncio.wait_for). - Test with AsyncClient in pytest and include async tests in CI.
- Monitor threadpool and event loop latencies in production (tools: Prometheus, Sentry with performance monitoring).
Final note — deployment & configuration
Uvicorn (or Gunicorn with uvicorn workers) runs the event loop. Use multiple worker processes when your application has CPU-bound parts; async concurrency helps with I/O-bound concurrency within each worker.
Pro tip: If you keep seeing
loop blocked for Xmswarnings or sudden latency spikes, you probably have blocking calls in your event loop — hunt them down like a caffeine-fueled wolf.
Wrap-up: what you should remember
async= coroutine function;await= pause here and let other work happen.- Async is for I/O. Sync is for stuff that can't be async (unless you offload it).
- Use async tests and include them in CI — you already set up CI for testing; now make sure async behavior is covered.
Go forth, convert your blocking spaghetti into beautiful awaitable lasagna, and when in doubt, add a timeout.
Version hint: next up, try pairing this with an async DB (e.g., encode/databases, SQLAlchemy 1.4+ async, or Tortoise) and write integration tests that run against a disposable DB in your CI pipeline.
Comments (0)
Please sign in to leave a comment.
No comments yet. Be the first to comment!