FastAPI Performance — Profiling, Caching and Query Optimisation

FastAPI can serve thousands of requests per second on a single server — but only if the code does not have performance bottlenecks. The most common bottlenecks in web applications are: N+1 database queries (loading 50 posts then querying the author for each one individually = 51 queries instead of 2), missing database indexes on filtered columns, and no caching on expensive read operations that are called thousands of times per minute with the same arguments.

Finding N+1 Queries #

# The N+1 problem — 1 query for posts + N queries for each author
@router.get("/posts")
def list_posts(db: Session = Depends(get_db)):
    posts = db.scalars(select(Post)).all()
    # When serialising: each post.author access fires a separate SELECT!
    return [PostResponse.model_validate(p) for p in posts]

# Fix: eager load authors with the initial query (2 queries total)
posts = db.scalars(
    select(Post)
    .options(selectinload(Post.author))   # ← load all authors in one query
    .options(selectinload(Post.tags))     # ← load all tags in one query
).all()

Redis Response Caching #

import redis.asyncio as aioredis
import json, hashlib

redis = aioredis.from_url(settings.redis_url)

def cache_key(prefix: str, **kwargs) -> str:
    """Generate a stable cache key from endpoint params."""
    param_str = json.dumps(kwargs, sort_keys=True)
    h         = hashlib.md5(param_str.encode()).hexdigest()[:8]
    return f"{prefix}:{h}"

@router.get("/posts", response_model=PaginatedPostsResponse)
async def list_posts(
    page:      int = 1,
    page_size: int = 10,
    tag:       str | None = None,
    db:        Session    = Depends(get_db),
):
    key    = cache_key("posts", page=page, page_size=page_size, tag=tag)
    cached = await redis.get(key)

    if cached:
        return json.loads(cached)   # ← cache hit, no DB query

    result = get_paginated_posts(db, page, page_size, tag)
    await redis.setex(key, 60, json.dumps(jsonable_encoder(result)))  # 60s TTL
    return result

# Invalidate cache when posts change
async def invalidate_posts_cache():
    keys = await redis.keys("posts:*")
    if keys: await redis.delete(*keys)

Database Indexes #

# In Alembic migration — add indexes for frequently filtered columns
def upgrade():
    # Posts filtered by status and ordered by created_at
    op.create_index("ix_posts_status_created",
                    "posts", ["status", "created_at"],
                    postgresql_ops={"created_at": "DESC"})

    # Posts full-text search on title
    op.execute("""
        CREATE INDEX ix_posts_title_fts
        ON posts
        USING gin(to_tsvector('english', title))
    """)

    # Post-tag association
    op.create_index("ix_post_tags_tag_id", "post_tags", ["tag_id"])

Common Mistakes #

Mistake 1 — Caching without invalidation (stale data) #

❌ Wrong — post list cached for 1 hour, new posts invisible for up to 1 hour.

✅ Correct — invalidate the post list cache on every create/update/delete operation.

Mistake 2 — joinedload on a collection with multiple relationships (Cartesian product) #

❌ Wrong — post with 5 tags and 5 comments returns 25 rows via joinedload:

select(Post).options(joinedload(Post.tags), joinedload(Post.comments))
# 5 × 5 = 25 rows returned, duplicated data!

✅ Correct — use selectinload for collections.