Why Async Matters for FastAPI
FastAPI is an ASGI framework - it can handle thousands of concurrent connections with a single process. But only if your database queries don’t block the event loop.
Time (seconds) 0────1────2────3────4────5
Sync (blocks): [───DB───] ← Thread blocked
[───DB───] ← Next request waits
[───DB───]
Async (non-blocking): [───DB───] ← Coroutine yields
[───DB───] ← Other requests interleave
[───DB───]
◄─── Event Loop runs ──►
| Approach | Concurrency | CPU Util | Requests/sec |
|---|---|---|---|
| Sync + thread pool | Thread-per-request | Medium | ~5,000 |
Async + await | Single-threaded cooperative | High | ~50,000 |
| Async + connection pool | Event loop + pooled DB | Very High | ~45,000 |
Sync database calls (e.g. psycopg2, pymysql) block the entire worker thread. Async drivers (asyncpg, aiomysql, aiosqlite) yield control back to the event loop while waiting for the database.
SQLAlchemy Async Session
The async version of SQLAlchemy uses AsyncSession and async_sessionmaker:
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost:5432/mydb"
engine = create_async_engine(DATABASE_URL, echo=True, pool_size=10, max_overflow=20)
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
Dependency Injection
from fastapi import Depends
async def get_db() -> AsyncSession:
async with async_session() as session:
yield session
@router.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
result = await db.execute(select(User).where(User.id == user_id))
user = result.scalar_one_or_none()
...
Depends(get_db)creates one session per request and closes it automatically when the request ends.
Awaiting Query Execution
Every SQLAlchemy async operation must be awaited:
# ── SELECT ──
result = await db.execute(select(User).where(User.email == email))
user = result.scalar_one_or_none()
# ── INSERT ──
db.add(new_user)
await db.commit()
await db.refresh(new_user)
# ── UPDATE ──
stmt = update(User).where(User.id == uid).values(name=new_name).returning(User)
result = await db.execute(stmt)
updated = result.scalar_one()
# ── DELETE ──
await db.execute(delete(User).where(User.id == uid))
await db.commit()
Streaming Large Results
For large result sets, use streaming to avoid loading everything into memory:
@router.get("/users/stream")
async def stream_users(db: AsyncSession = Depends(get_db)):
result = await db.stream(select(User))
users = []
async for row in result:
users.append(row.User)
return users
Multiple Queries in Parallel
Use asyncio.gather to run independent queries concurrently:
from asyncio import gather
@router.get("/dashboard")
async def get_dashboard(db: AsyncSession = Depends(get_db)):
async def count_users():
result = await db.execute(select(func.count(User.id)))
return result.scalar()
async def count_posts():
result = await db.execute(select(func.count(Post.id)))
return result.scalar()
async def recent_posts():
result = await db.execute(select(Post).order_by(Post.created_at.desc()).limit(5))
return result.scalars().all()
user_count, post_count, latest_posts = await gather(
count_users(), count_posts(), recent_posts()
)
return {
"users": user_count,
"posts": post_count,
"recent": latest_posts,
}
Connection Pooling
SQLAlchemy’s connection pool reuses database connections instead of opening a new one per request:
engine = create_async_engine(
DATABASE_URL,
pool_size=10, # Persistent connections
max_overflow=20, # Temporary extra connections under load
pool_timeout=30, # Seconds to wait for a connection from pool
pool_recycle=1800, # Recycle connections after 30 minutes
pool_pre_ping=True, # Verify connection before using it
)
Pool Behavior Under Load
Requests ──► Pool ──► Database
┌───┐
Req 1 ──────────►│ C1 │──► DB
Req 2 ──────────►│ C2 │──► DB
Req 3 ──────────►│ C3 │──► DB
...
Req 11 ─────────►│ O1 │──► DB (max_overflow)
└───┘
Req 21 ──► waits for pool_timeout ──► Error
When all pool + overflow connections are in use, new requests wait (up to pool_timeout seconds) and then raise TimeoutError. Size your pool to handle peak concurrency.
Disabling Pooling (for serverless)
engine = create_async_engine(
DATABASE_URL,
poolclass=NullPool, # Create a new connection per request
)
Serverless environments (AWS Lambda, Cloud Functions) should use NullPool or a connection pooler like PgBouncer.
The N+1 Problem
N+1 happens when you load a parent object, then loop over its children - issuing one query for the parent and N queries for N children.
The Problem
# BAD: 1 query for posts + N queries for authors
@router.get("/posts")
async def list_posts(db: AsyncSession = Depends(get_db)):
result = await db.execute(select(Post))
posts = result.scalars().all()
output = []
for post in posts:
# One await per post! N queries!
result = await db.execute(select(User).where(User.id == post.author_id))
author = result.scalar_one()
output.append({"title": post.title, "author": author.name})
return output
With 100 posts, this issues 101 queries.
The Fix: Eager Loading with selectinload
from sqlalchemy.orm import selectinload
@router.get("/posts")
async def list_posts(db: AsyncSession = Depends(get_db)):
result = await db.execute(
select(Post).options(selectinload(Post.author))
)
posts = result.scalars().all()
return [
{"title": p.title, "author": p.author.name}
for p in posts
]
This issues 2 queries total - one for posts, one for all related authors via WHERE id IN (...).
Loading Strategies
| Strategy | Behavior | Best For |
|---|---|---|
selectinload | Eager load via SELECT ... WHERE id IN (...) | One-to-many, many-to-many |
joinedload | Eager load via LEFT JOIN | Many-to-one, one-to-one |
lazyload | Load on access (triggers N+1) | Avoid unless you know what you are doing |
subqueryload | Eager load via subquery | Complex hierarchies |
# Many-to-one: use joinedload (single JOIN is cheaper)
result = await db.execute(
select(Post).options(joinedload(Post.author))
)
# One-to-many: use selectinload (avoids Cartesian product)
result = await db.execute(
select(User).options(selectinload(User.posts))
)
Async vs Sync at a Glance
Sync Async
┌─────────────────┐ ┌─────────────────┐
│ def get_user(): │ │ async def get(): │
│ user = db.query │ │ await db.execute │
│ return user │ │ return user │
└────────┬────────┘ └────────┬────────┘
│ │
Blocks thread ◄────── await ──────► Yields to loop
│ │
Other requests Other requests
wait in queue ◄─────── and ──────► interleave freely
| Aspect | Sync | Async |
|---|---|---|
| Thread model | Blocks worker thread | Yields event loop |
| Concurrency | Thread pool (limited) | Cooperative (thousands) |
| Driver | psycopg2, pymysql | asyncpg, aiomysql, aiosqlite |
| SQLAlchemy | Session (sync) | AsyncSession (async) |
| Code | db.query(User).all() | await db.execute(select(User)) |
| Caveats | Thread pool exhaustion | Must await everything |
| Serverless | Better cold start (no event loop overhead) | Better throughput under concurrency |
Rule of thumb: If your FastAPI app handles I/O-bound workloads (DB calls, HTTP requests, file uploads), use async every time. If you’re CPU-bound (ML inference, image processing), sync is fine - offload to a thread pool.
Key Takeaways
- Async prevents the event loop from blocking during database I/O - critical for FastAPI’s concurrency model
- Use
AsyncSessionfrom SQLAlchemy’sext.asynciomodule; every DB call must beawaited - Pool connections with
pool_sizeandmax_overflow; useNullPoolfor serverless environments - Use
asyncio.gatherto run independent queries in parallel for dashboards and complex endpoints - Avoid N+1 - use
selectinloadfor one-to-many andjoinedloadfor many-to-one relationships - Sync queries block the thread and limit concurrency; async queries yield the event loop for higher throughput
- Streaming large result sets with
async forprevents memory exhaustion