Use Case 7: Caching for High-Performance Reads
Scenario: A read-heavy API (product catalog) reduces DB load by caching documents by _id — in-memory locally, or shared via Redis.
📦 What's New?
| Component | Description |
|---|---|
InMemoryCacheBackend |
TTL + LRU cache with a background cleanup task. |
RedisCacheBackend |
Distributed cache on redis.asyncio with pub/sub invalidation. |
CachedBaseRepository[T] |
Extends BaseRepository — cache-first get_by_id, cache on create, invalidate on update/delete, warm_cache(ids), invalidate_cache(id). |
| Backend lifecycle | The same backend instance must be both passed to the repository AND registered via ModelRegistry.set_cache_backend so initialize_cache() starts its task. |
🚀 Example
Python
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from mongo_ops import BaseDocument, CachedBaseRepository, ModelRegistry, MongoConnectionManager
from mongo_ops.cache import CacheConfig, InMemoryCacheBackend
class Product(BaseDocument):
name: str = ""
price: float = 0.0
# One shared backend — used by both the repository and the registry lifecycle.
cache = InMemoryCacheBackend(max_entries=10_000, default_ttl=300)
ModelRegistry.set_cache_backend(cache)
class ProductRepo(CachedBaseRepository[Product]):
def __init__(self):
super().__init__(
collection_name="products",
model=Product,
cache_backend=cache,
config=CacheConfig(enabled=True, backend="memory"),
)
@asynccontextmanager
async def lifespan(_app: FastAPI):
async with MongoConnectionManager.lifespan(
uri="mongodb://localhost:27017", db_name="shop"
):
await ModelRegistry.initialize_all()
await ModelRegistry.initialize_cache() # starts the TTL cleanup task
yield
await ModelRegistry.shutdown_cache() # cancels it on exit
app = FastAPI(lifespan=lifespan)
@app.post("/products/", response_model=Product)
async def create_product(product: Product):
return await ProductRepo().create(product) # created after connect()
@app.get("/products/{product_id}", response_model=Product)
async def get_product(product_id: str):
product = await ProductRepo().get_by_id(product_id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
return product
@app.put("/products/{product_id}", response_model=Product)
async def update_product(product_id: str, name: str | None = None, price: float | None = None):
data = {}
if name is not None:
data["name"] = name
if price is not None:
data["price"] = price
return await ProductRepo().update(product_id, data)
# Outside request handlers:
# await ProductRepo().warm_cache([object_id_1, object_id_2]) # pre-load n ids -> int
# await ProductRepo().invalidate_cache(object_id_3) # manual eviction
🔄 Redis Backend
Swap the backend — the repository code stays identical:
Python
from mongo_ops.cache.redis_backend import RedisCacheBackend
from redis.asyncio import Redis
redis_client = Redis(host="localhost", port=6379)
redis_backend = RedisCacheBackend(redis_client, key_prefix="prod:")
ModelRegistry.set_cache_backend(redis_backend) # for the lifecycle in lifespan()
class ProductRepo(CachedBaseRepository[Product]):
def __init__(self):
super().__init__(
collection_name="products",
model=Product,
cache_backend=redis_backend,
config=CacheConfig(enabled=True, backend="redis"),
)
💡 Tips
- Cache keys are
"{key_prefix}{id}"(default prefix"products:").clear_pattern("products:*")wipes a whole collection's entries. - Values are JSON-encoded (
json.dumps(obj, default=str)) — nested models inside a cached document are stored as dicts, not objects. - Enable/disable per repository with
CacheConfig(enabled=False); a disabled repo bypasses the cache entirely. - Distinguish the two
initialize*calls:initialize_cache()starts the backend task;initialize_all()creates indexes. Both belong in the lifespan, afterconnect().