Production API with HTTP Caching

You have a FastAPI endpoint serving product catalog data. It hits a database, does some computation, and returns JSON. Traffic is growing and you need to reduce database load and response times.

This recipe combines every layer inhouse offers: in-process caching, HTTP Cache-Control for CDN offload, ETag/304 for bandwidth savings, parameter exclusion for clean keys, and surgical invalidation when data changes.

The catalog endpoint

from fastapi import FastAPI, Request
from inhouse import MemoryStore
from inhouse.fastapi import create_lifespan, fastapi_cache

store = MemoryStore(max_size=2048, default_ttl=300)
app = FastAPI(lifespan=create_lifespan(store))

@app.get("/catalog/{category}")
@fastapi_cache(
    300,
    store=store,
    http_cache=True,
    etag=True,
    cache_visibility="public",
)
async def get_catalog(category: str, sort: str = "name") -> list[dict]:
    rows = await db.fetch_all(
        "SELECT * FROM products WHERE category = $1 ORDER BY $2",
        category, sort,
    )
    return [dict(r) for r in rows]

Three layers active:

• In-process cache: same (category, sort) combo returns instantly after the first hit
• Cache-Control: public, max-age=300: CDN and browsers won't even send the request for 5 minutes
• ETag: if a conditional request arrives (after expiry), a 304 Not Modified skips the response body

Drop unnecessary parameters from the cache key

Your endpoint signature grows a request parameter for analytics. Without exclusion, every request gets a different cache key.

@app.get("/catalog/{category}")
@fastapi_cache(
    300,
    store=store,
    http_cache=True,
    etag=True,
    cache_visibility="public",
    exclude=("request",),
)
async def get_catalog(
    request: Request,  # for analytics - not part of cache key
    category: str,
    sort: str = "name",
) -> list[dict]:
    track_request(request)
    rows = await db.fetch_all(...)
    return [dict(r) for r in rows]

request is excluded from the key. get_catalog("electronics", sort="price") produces the same key regardless of which user or device made the request.

Surgical invalidation after data changes

An admin panel updates product prices. The catalog cache needs to refresh for the affected category only.

from pydantic import BaseModel

class PriceUpdate(BaseModel):
    product_id: int
    category: str
    new_price: float

@app.post("/catalog/{category}/update-price")
async def update_price(update: PriceUpdate):
    await db.execute(
        "UPDATE products SET price = $1 WHERE id = $2",
        update.new_price, update.product_id,
    )
    # Evict only this category's cached catalog
    get_catalog.cache_invalidate(
        request=None,  # excluded anyway, but required by signature
        category=update.category,
        sort="name",
    )
    return {"ok": True}

The eviction is per-category. If an admin updates a product in "electronics", only get_catalog("electronics", "name") is evicted. "home-garden" is untouched.

Use keyword arguments matching the route signature. cache_invalidate builds the key the same way the decorator does.

Monitoring cache performance

@app.get("/admin/cache-stats")
async def cache_stats():
    s = store.stats()
    total = s["hits"] + s["misses"]
    hit_ratio = s["hits"] / total if total else 0.0
    return {
        "hit_ratio": round(hit_ratio, 3),
        "entries": s["size"],
        "evictions": s["evictions"],
        "sets": s["sets"],
    }

Check this endpoint after deployment to confirm the cache is working. High evictions mean max_size is too low. Low hit ratio means TTL is too short or keys are too specific.

Handling cache independence

In-process TTL and HTTP Cache-Control are independent. When a price update triggers cache_invalidate, the in-process entry is deleted. But a CDN or browser that received Cache-Control: public, max-age=300 still has a stale copy until that 300 seconds passes.

@app.post("/catalog/{category}/update-price")
async def update_price(update: PriceUpdate):
    await db.execute(...)

    # Evict in-process cache
    removed = get_catalog.cache_invalidate(
        request=None,
        category=update.category,
        sort="name",
    )

    # Tell CDN the resource changed (if you control the CDN API)
    await purge_cdn_cache(f"/catalog/{update.category}")

    return {"ok": True, "cache_evicted": removed}

For full control, use a short max-age (60-120 seconds) so CDN copies expire quickly, and rely on in-process caching for the fast path within that window.

Putting it all together

from fastapi import FastAPI, Request
from inhouse import MemoryStore
from inhouse.fastapi import create_lifespan, fastapi_cache

store = MemoryStore(max_size=4096, default_ttl=120)
app = FastAPI(lifespan=create_lifespan(store))

@app.get("/api/v2/products/{product_id}")
@fastapi_cache(
    120,
    store=store,
    http_cache=True,
    etag=True,
    cache_visibility="public",
    exclude=("request",),
)
async def get_product(request: Request, product_id: int) -> dict | None:
    row = await db.fetch_row(
        "SELECT * FROM products WHERE id = $1", product_id
    )
    return dict(row) if row else None

@app.get("/api/v2/products")
@fastapi_cache(
    60,
    store=store,
    http_cache=True,
    etag=True,
    cache_visibility="public",
    exclude=("request",),
)
async def list_products(
    request: Request,
    category: str | None = None,
    sort: str = "name",
    page: int = 1,
) -> list[dict]:
    ...

@app.post("/api/v2/products/{product_id}")
async def update_product(product_id: int, data: ProductUpdate):
    await db.execute("UPDATE products SET ... WHERE id = $1", product_id)
    get_product.cache_invalidate(request=None, product_id=product_id)
    list_products.cache_clear()  # listing may have changed
    return {"ok": True}

Product detail pages are cached for 2 minutes with CDN offload and ETag support. Listings are cached for 1 minute. Writes evict the affected product's detail entry and clear the listing cache. Stats are available from the store at any time.