Response cache¶
A shop's catalogue and account pages behind @cache: product pages are
the same for everybody and share one entry each, account pages depend on who asks and get entries
of their own, an update drops the entries of a product by tag,
and get_or_set caches what
doesn't fit a decorator. Every entry is signed, so whoever can write to Redis can't plant one.
Run it¶
From a clone of the repository:
make deps # .venv, with every extra
source .venv/bin/activate
uvicorn examples.cache.main:app
It keeps the entries in memory. To share them between processes, give it a Redis or Valkey and a signing key, the same in every process:
export REDIS_URL=redis://localhost:6379/0
export CACHE_SIGNING_KEY=$(python -c "import secrets; print(secrets.token_urlsafe(32))")
uvicorn examples.cache.main:app --workers 4
Try it¶
The first request computes the page and stores it, the next ones get it from the cache, as the
Cache-Status header says, with the seconds the entry
has left:
$ curl -i http://127.0.0.1:8000/products/1
HTTP/1.1 200 OK
etag: W/"8a7a39cf30d133a472a8c4f0beca023d"
cache-status: Alcazaba; fwd=miss; stored
cache-control: max-age=600
{"id":1,"name":"Olive oil","price":9.5}
$ curl -i http://127.0.0.1:8000/products/1
HTTP/1.1 200 OK
etag: W/"8a7a39cf30d133a472a8c4f0beca023d"
cache-status: Alcazaba; hit; ttl=600
cache-control: max-age=600
{"id":1,"name":"Olive oil","price":9.5}
A personal page is cached per Authorization header and marked private, so neither another user
nor a shared cache in between gets it:
$ curl -i http://127.0.0.1:8000/me -H "Authorization: Bearer alice"
HTTP/1.1 200 OK
cache-status: Alcazaba; fwd=miss; stored
cache-control: private, max-age=30
vary: Authorization, Cookie
{"user":"alice"}
An update drops the product's entry, whatever its key:
$ curl -X PUT http://127.0.0.1:8000/products/1 -H "Content-Type: application/json" \
-d '{"id": 1, "name": "Olive oil", "price": 10}'
$ curl -i http://127.0.0.1:8000/products/1
HTTP/1.1 200 OK
cache-status: Alcazaba; fwd=miss; stored
{"id":1,"name":"Olive oil","price":10.0}
What to look at¶
vary_headers=()only for public data: by default a request withAuthorizationorCookiegets an entry of its own, sent asCache-Control: private. The product page opts out because it's the same for everybody; the account page keeps the default.- Tags instead of keys: the product page is tagged
product:<id>, so the update drops it (and anything else with that tag) without rebuilding its key. - Errors aren't cached: the 404 of an unknown product is computed every time, never replayed.
get_or_setfor the rest: the recommendations are cached under a key the code chooses, through the same pipeline as the decorator (signing, single flight, locks).- A signing key from the environment: the fallback in the code is only for trying it out. With Redis, every process must share the same key, or each refuses the others' entries.
The code¶
examples/cache/main.py
"""A shop's catalogue and account pages, cached by Alcazaba.
uvicorn examples.cache.main:app
With REDIS_URL (redis://localhost:6379/0) the entries live in Redis and every process shares
them; without it, in this process's memory. Either way they're signed: set CACHE_SIGNING_KEY
(secrets.token_urlsafe(32)), the same in every process.
- `/products/{product_id}`: the same for everybody, one shared entry per product, tagged so that
an update drops it (and the lists it appears in) without knowing their keys.
- `/me`: depends on who asks. Requests with `Authorization` get entries of their own, sent as
`Cache-Control: private`, so no user ever gets another's page.
- `/me/recommendations`: `get_or_set`, for what doesn't fit a decorator.
- `exchange_rate`: a plain function cached like an endpoint.
"""
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Annotated
import uvicorn
from fastapi import Depends, FastAPI, Header, HTTPException
from pydantic import BaseModel, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict
from fastapi_alcazaba import Cache, FastAPICache, get_cache
from fastapi_alcazaba.backends import Backend
from fastapi_alcazaba.backends.inmemory import InMemoryBackend
from fastapi_alcazaba.backends.redis import RedisBackend
from fastapi_alcazaba.response_cache import cache
from redis import asyncio as aioredis
class Product(BaseModel):
id: int
name: str
price: float
class Shop:
"""Stands in for the database, counting the queries that reach it."""
def __init__(self) -> None:
self.products = {
1: Product(id=1, name="Olive oil", price=9.5),
2: Product(id=2, name="Olives", price=3.2),
}
self.queries = 0
def product(self, product_id: int) -> Product | None:
self.queries += 1
return self.products.get(product_id)
def recommendations(self, _user: str) -> list[Product]:
"""The same for everybody here; a real shop would look at the user's history."""
self.queries += 1
return list(self.products.values())
shop = Shop()
class Settings(BaseSettings):
"""The configuration, from environment variables or a .env file."""
model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")
redis_url: str | None = None # e.g. redis://localhost:6379/0; without it, in memory
# Throwaway key for trying it out: a real one comes from the environment or a secret
# manager, and is the same in every process.
cache_signing_key: SecretStr = SecretStr("development-only-signing-key-0123456")
settings = Settings()
@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncIterator[None]:
url = settings.redis_url
client = aioredis.from_url(url) if url else None
backend: Backend = RedisBackend(client) if client is not None else InMemoryBackend()
FastAPICache.init(
backend,
prefix="shop",
signing_key=settings.cache_signing_key,
)
yield
if client is not None:
await client.aclose()
app = FastAPI(lifespan=lifespan)
def current_user(authorization: Annotated[str | None, Header()] = None) -> str:
"""Stands in for real authentication: the token is the user's name."""
if authorization is None:
raise HTTPException(401, "Not authenticated", headers={"WWW-Authenticate": "Bearer"})
return authorization.removeprefix("Bearer ")
@app.get("/products/{product_id}")
@cache(expire=600, vary_headers=(), tags=lambda product_id, **_: [f"product:{product_id}"])
async def product(product_id: int) -> Product:
"""Public data: one entry per product, whoever asks (`vary_headers=()`)."""
found = shop.product(product_id)
if found is None:
raise HTTPException(404, "No such product") # errors are never cached
return found
@app.put("/products/{product_id}")
async def update_product(product_id: int, update: Product) -> Product:
shop.products[product_id] = update
await FastAPICache.invalidate_tags(f"product:{product_id}")
return update
@app.get("/me")
@cache(expire=30)
async def me(user: Annotated[str, Depends(current_user)]) -> dict[str, str]:
"""Personal data: an entry per Authorization header, never shared."""
return {"user": user}
@app.get("/me/recommendations")
async def recommendations(
user: Annotated[str, Depends(current_user)],
app_cache: Annotated[Cache, Depends(get_cache)],
) -> list[Product]:
async def load() -> list[Product]: # annotated: the entry comes back as products
return shop.recommendations(user)
return await app_cache.get_or_set(f"recommendations:{user}", load, expire=300)
@cache(expire=3600)
async def exchange_rate(currency: str) -> float:
"""A slow call to another service, made at most once an hour per currency."""
return {"USD": 1.08, "GBP": 0.86}.get(currency, 1.0)
@app.get("/products/{product_id}/price")
async def price(product_id: int, currency: str = "EUR") -> dict[str, float]:
found = shop.product(product_id)
if found is None:
raise HTTPException(404, "No such product")
return {"price": round(found.price * await exchange_rate(currency), 2)}
if __name__ == "__main__":
uvicorn.run("main:app", reload=True)
The repository also has examples/redis,
with Redis, signed entries, a sync endpoint run in the thread pool and an HTML page cached with
PickleCoder.