KAHIBARO
Discord Login Register ☰
☰

Caching Database Queries

Table of Contents

Why Cache Database Queries?

When your backend grows, database queries often become the slowest part of each request. Every time an API endpoint needs data, it talks to the database, the database reads from disk or memory, runs the query, and sends results back.

If many users request the same data repeatedly, this becomes wasteful. Caching database queries lets you:

In this chapter you will see how to cache query results, when it helps, and how to avoid common mistakes.

Key idea: Cache database query results that are:

  • Expensive to compute,
  • Read frequently,
  • Do not change very often.

Types of Data That Benefit from Query Caching

Not all queries are good candidates for caching. Some are better than others.

Good candidates for caching

These queries are usually excellent to cache:

Poor candidates for caching

Some queries should usually not be cached, or only very briefly:

In these cases, caching can introduce bugs if the cache is slightly out of date.

Rule of thumb

If running a query often is expensive and the answer does not change every millisecond, you can usually cache it for at least a short time.

Basic Pattern: Read-Through Cache for Queries

The classic pattern for caching a query result is called read through.

Conceptually, you do this:

  1. Look in the cache for the query result.
  2. If present, return it.
  3. If not present, run the query against the database.
  4. Store the result in the cache.
  5. Return the result.

In pseudocode:

python 
def get_something(params):
    key = make_cache_key(params)
    cached = cache.get(key)
    if cached is not None:
        return cached
    # Cache miss
    data = run_query(params)
    cache.set(key, data, ttl=60)  # store for 60 seconds
    return data

You will see real examples shortly. First, you need to understand how to build a good cache key.

Designing Cache Keys for Queries

A cache key identifies a cached value. For database query caching, the key usually represents:

Rule: If two different requests could return different query results, they must use different cache keys.

Components of a query cache key

Common parts of a key:

Example structure:

text 
"{prefix}:{resource}:{params_hash}"

or

text 
"{prefix}:{resource}:{param1}:{param2}:{...}"

Simple key example

Imagine an endpoint:

GET /products?category=books&page=2&page_size=20&sort=price_asc

A possible key:

text 
"products:list:category=books:page=2:page_size=20:sort=price_asc"

In Python:

python 
def make_products_list_key(category, page, page_size, sort):
    return f"products:list:category={category}:page={page}:size={page_size}:sort={sort}"

Keys with many parameters

If your query has many parameters, long keys can become messy. In these cases, you can:

  1. Serialize parameters as JSON.
  2. Hash the JSON to create a compact string.

Example:

python 
import json
import hashlib
def make_cache_key(prefix: str, **params) -> str:
    # Sort keys to keep order stable
    params_json = json.dumps(params, sort_keys=True)
    params_hash = hashlib.sha256(params_json.encode()).hexdigest()
    return f"{prefix}:{params_hash}"

Usage:

python 
key = make_cache_key(
    "products:list",
    category="books",
    page=2,
    page_size=20,
    sort="price_asc",
)

The actual key looks like:

text 
"products:list:8a2a9c1d8b7f..."

Easy to store, stable, and unique for each parameter combination.

Where to Cache: Per Query or Per Entity?

You can cache at different levels.

1. Cache whole query results

Cache the entire result set for a query, usually as a JSON blob.

Example:

python 
def get_products_list(category, page, page_size, sort):
    key = make_products_list_key(category, page, page_size, sort)
    cached = redis.get(key)
    if cached:
        return json.loads(cached)
    # Run database query
    products = db.fetch_products(category, page, page_size, sort)
    redis.setex(key, 60, json.dumps(products))  # 60-second cache
    return products

Pros:

Cons:

2. Cache individual entities by ID

Sometimes you cache each row separately, for example by user:{id} or product:{id}.

Example:

python 
def get_product_by_id(product_id: int):
    key = f"product:{product_id}"
    cached = redis.get(key)
    if cached:
        return json.loads(cached)
    product = db.fetch_product_by_id(product_id)
    if product is None:
        return None
    redis.setex(key, 300, json.dumps(product))
    return product

Pros:

Cons:

3. Combine both levels

Common real world approach:

This gives more flexibility in invalidation and reuse.

Serialization: Storing Query Results in the Cache

Most caches, like Redis, store strings or binary blobs. Database query results in Python are often:

You must serialize these objects.

Common formats

FormatProsCons
JSONHuman readable, portable, language independentOnly basic types, slow for very large objects
PickleSimple for Python objectsPython only, security risk if misused
MessagePackCompact, faster than JSONNeeds extra library

For APIs, JSON is usually the safest choice.

Example: JSON serialization in Python

python 
import json
def cache_set_json(key: str, value, ttl: int | None = None):
    data = json.dumps(value)
    if ttl:
        redis.setex(key, ttl, data)
    else:
        redis.set(key, data)
def cache_get_json(key: str):
    data = redis.get(key)
    if data is None:
        return None
    return json.loads(data)

Use these helpers in your query caching code:

python 
def get_top_posts(limit: int = 10):
    key = f"posts:top:{limit}"
    cached = cache_get_json(key)
    if cached is not None:
        return cached
    posts = db.fetch_top_posts(limit=limit)
    cache_set_json(key, posts, ttl=120)
    return posts

Choosing a Time To Live (TTL) for Query Caches

The TTL decides how long a cached value stays valid. When TTL expires, the cache entry disappears and you have to fetch fresh data from the database.

There is a trade off:

Practical TTL guidelines

Different queries may use different TTL values. Here are some rough ideas:

Data typeExampleSuggested TTL
Rarely changing reference dataCountry list1 hour to 24 hours
Homepage / landing page listsLatest articles30 to 300 seconds
Trending / top listsTop posts, top products30 to 300 seconds
Dashboard statisticsDaily summary60 to 600 seconds
Per user light personalization"Recently viewed"30 to 120 seconds

This is not strict. You adjust based on your system.

Use different TTLs in the same app

Your application can easily apply different TTLs:

python 
TTL = {
    "reference": 3600,
    "homepage": 120,
    "top_list": 180,
    "user_data": 60,
}

Then:

python 
cache_set_json(key, value, ttl=TTL["top_list"])

Cache Invalidation for Database Queries

The tricky part of caching is not storing or reading values. It is invalidation. Invalidation means making sure the cache does not serve outdated results when the underlying database changes.

Rule: Any time data changes in the database in a way that affects a cached query, the cache must be invalidated or refreshed, or the TTL must be short enough that staleness is acceptable.

There are two main strategies:

  1. Passive invalidation with TTLs.
  2. Active invalidation when data changes.

You can also combine both.

1. Passive invalidation (rely on TTLs)

In this approach:

Example:

python 
def get_latest_articles():
    key = "articles:latest"
    cached = cache_get_json(key)
    if cached is not None:
        return cached
    articles = db.fetch_latest_articles(limit=10)
    cache_set_json(key, articles, ttl=60)
    return articles

If a new article is added, the cache might still serve the old list for up to 60 seconds. For many use cases, this is acceptable.

Pros:

Cons:

2. Active invalidation on writes

In this approach, whenever something changes in the database, you actively remove or update cache entries that depend on that data.

For example:

Basic pattern:

python 
def update_product(product_id: int, updates: dict):
    product = db.update_product(product_id, updates)
    # Invalidate caches
    redis.delete(f"product:{product_id}")
    redis.delete("products:top:10")
    redis.delete("products:latest:10")
    # etc.
    return product

Pros:

Cons:

Using key prefixes for easier invalidation

Instead of listing every key, you can use key prefixes and delete all keys with a given prefix. For Redis you often use SCAN plus DEL.

Example structure:

To invalidate all product lists:

python 
def invalidate_product_lists():
    pattern = "product:list:*"
    cursor = 0
    while True:
        cursor, keys = redis.scan(cursor=cursor, match=pattern, count=100)
        if keys:
            redis.delete(*keys)
        if cursor == 0:
            break

However, scanning in production must be used carefully and not too often. It can be slow if you have many keys.

Combining Entity Cache and Query Cache

Suppose you have:

One robust pattern:

  1. Cache each product by ID:
    • Key: product:{id}.
  2. Cache each list as a list of IDs only:
    • Key: product:list:category=books:page=1:....
  3. When serving a list:
    • Get list of product IDs from list cache.
    • For each ID, fetch product from product:{id}, then combine into a list.

Example

python 
def get_product_by_id(product_id: int):
    key = f"product:{product_id}"
    cached = cache_get_json(key)
    if cached is not None:
        return cached
    product = db.fetch_product_by_id(product_id)
    if not product:
        return None
    cache_set_json(key, product, ttl=300)
    return product
def get_products_list(category: str, page: int, page_size: int):
    list_key = make_products_list_key(category, page, page_size)
    cached_ids = cache_get_json(list_key)
    if cached_ids is not None:
        products = []
        for pid in cached_ids:
            product = get_product_by_id(pid)
            if product:
                products.append(product)
        return products
    # Cache miss for list of IDs
    products = db.fetch_products(category, page, page_size)
    ids = [p["id"] for p in products]
    cache_set_json(list_key, ids, ttl=60)
    # Also warm up entity cache
    for p in products:
        cache_set_json(f"product:{p['id']}", p, ttl=300)
    return products

Benefits:

Example: Caching a Frequently Used Report

Imagine an endpoint for a dashboard:

GET /reports/orders-summary?month=2024-01

The database query calculates:

The query touches large tables and is slow, about 1 second.

Step 1: Define a cache key

python 
def make_orders_summary_key(month: str) -> str:
    return f"reports:orders_summary:month={month}"

Step 2: Implement caching

python 
def get_orders_summary(month: str):
    key = make_orders_summary_key(month)
    cached = cache_get_json(key)
    if cached is not None:
        return cached
    # Slow query
    summary = db.fetch_orders_summary(month=month)
    # Cache for 10 minutes
    cache_set_json(key, summary, ttl=600)
    return summary

This is usually enough because summaries do not change often once the month is complete.

Step 3: Handle updates

If your system allows modifying orders of past months, you have two options:

  1. Short TTL
    Keep TTL small, for example 60 seconds, and accept minor delay.
  2. Invalidate when updating orders
    When an order changes month or amount:
python 
   def update_order(order_id: int, updates: dict):
       old_order = db.fetch_order(order_id)
       order = db.update_order(order_id, updates)
       # Invalidate affected months
       if "month" in updates:
           old_month = old_order["month"]
           new_month = order["month"]
           redis.delete(make_orders_summary_key(old_month))
           redis.delete(make_orders_summary_key(new_month))
       else:
           month = order["month"]
           redis.delete(make_orders_summary_key(month))
       return order

Now your monthly summary cache always becomes correct after any update.

Avoiding Cache Stampede

A cache stampede happens when:

Example scenario

This defeats the purpose of caching.

Simple strategies to reduce stampedes

You can use several techniques without complex tools.

1. Staggered or random TTL

Instead of a fixed TTL like 60 seconds, use a small random variation, for example 50 to 70 seconds. This way, not all keys expire at the exact same moment.

python 
import random
def set_with_jitter(key, value, base_ttl: int):
    jitter = random.randint(-10, 10)  # from -10 to +10 seconds
    ttl = max(10, base_ttl + jitter)
    cache_set_json(key, value, ttl=ttl)

Use it:

python 
set_with_jitter(key, value, base_ttl=60)
2. Soft TTL and background refresh

You can combine:

Very simple version:

Full implementation is more advanced, but the idea is:

  1. Store value and expires_at in the cache.
  2. If current time is before expires_at, serve from cache.
  3. If close to expires_at, serve from cache but also start a background job to refresh.

This reduces many simultaneous database hits.

Caching with ORM Queries

When you use an ORM like SQLAlchemy, you usually work with model objects, not raw SQL. You can still apply exactly the same caching patterns, but you must serialize ORM objects to JSON or dictionaries first.

Converting ORM objects

Suppose you have a Product model with a .to_dict() method:

python 
class Product(Base):
    __tablename__ = "products"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    price = Column(Numeric)
    def to_dict(self):
        return {
            "id": self.id,
            "name": self.name,
            "price": float(self.price),
        }

Then your caching function becomes:

python 
def get_product_by_id(session, product_id: int):
    key = f"product:{product_id}"
    cached = cache_get_json(key)
    if cached is not None:
        return cached
    product = session.query(Product).get(product_id)
    if not product:
        return None
    data = product.to_dict()
    cache_set_json(key, data, ttl=300)
    return data

For lists:

python 
def get_products_list(session, category: str, page: int, page_size: int):
    key = make_products_list_key(category, page, page_size)
    cached = cache_get_json(key)
    if cached is not None:
        return cached
    query = (
        session.query(Product)
        .filter(Product.category == category)
        .order_by(Product.id)
        .offset((page - 1) * page_size)
        .limit(page_size)
    )
    products = [p.to_dict() for p in query.all()]
    cache_set_json(key, products, ttl=60)
    return products

The database and ORM work exactly as before. The cache sits in front and stores serialized results.

Measuring Benefits of Query Caching

You should not guess whether query caching helps. You can measure it.

Useful metrics

Track:

For example, you can add simple logging:

python 
def get_with_cache(key, loader, ttl: int):
    value = cache_get_json(key)
    if value is not None:
        log.info("cache_hit", key=key)
        return value
    log.info("cache_miss", key=key)
    value = loader()
    cache_set_json(key, value, ttl=ttl)
    return value

Then use:

python 
def get_latest_articles():
    key = "articles:latest"
    return get_with_cache(key, loader=db.fetch_latest_articles, ttl=60)

Over time, you can see how often your cached queries avoid hitting the database.

Common Pitfalls and How to Avoid Them

1. Forgetting to include parameters in the key

If you accidentally ignore a parameter when constructing cache keys, different queries may share the same cached result.

Example mistake:

python 
# Bug: missing "sort" in the key
def make_key(category, page, page_size, sort):
    return f"products:list:category={category}:page={page}:size={page_size}"

Then requests with sort=price_asc and sort=price_desc share the same cache entry and return incorrect data.

Always ensure every parameter that can change the result is reflected in the key.

2. Caching non deterministic queries

Do not cache queries that are random by definition, for example:

sql 
SELECT * FROM posts ORDER BY RANDOM() LIMIT 5;

If you cache this for 1 hour, "random" will look very fixed. If you want fresh randomness, you may choose not to cache or to cache for a very short time.

3. Caching very large result sets

If a query returns huge lists (for example tens of thousands of rows) and you try to cache the entire list as JSON, you:

In such cases:

4. Not handling "not found" results

If an item is not found in the database, you might want to cache that negative result for a short time, particularly if that lookup is expensive.

Example, caching None:

python 
def get_user_by_email(email: str):
    key = f"user:by_email:{email}"
    cached = cache_get_json(key)
    if cached is not None:
        # We store {"found": False} or similar
        if not cached["found"]:
            return None
        return cached["user"]
    user = db.fetch_user_by_email(email)
    if not user:
        cache_set_json(key, {"found": False}, ttl=60)
        return None
    cache_set_json(key, {"found": True, "user": user}, ttl=300)
    return user

This avoids repeated expensive lookups for missing users.

Summary

Caching database queries can dramatically improve your backend performance and relieve pressure on your database. The essential ideas are:

With these patterns you can confidently apply caching to database queries in your own backend projects and understand the trade offs you are making.

Views: 9

Comments

Please login to add a comment.

Don't have an account? Register now!

© 2025 Kahibaro
Visitors: 3105 / 24h
Impressum / Imprint
Datenschutz / Data Policy