Cache Strategy

Cache Strategy — Architecture Patterns

Use this skill for cache architecture decisions. Covers Redis patterns, CDN strategies, TTL management, cache invalidation, and session stores. Used by Demeter during schema design and Hermes during implementation.

Cache Patterns

1. Read-Through Cache

The cache sits between the application and the database. On a cache miss, the cache loads the data from the database.

import redis
import json
from typing import Optional

class ReadThroughCache:
    def __init__(self, redis_client: redis.Redis, ttl: int = 300):
        self.redis = redis_client
        self.ttl = ttl
    
    async def get(self, key: str, loader) -> Optional[dict]:
        """Get from cache. If miss, load from source and cache."""
        cached = await self.redis.get(key)
        if cached:
            return json.loads(cached)
        
        # Cache miss — load from source
        data = await loader()
        if data:
            await self.redis.setex(key, self.ttl, json.dumps(data))
        return data

When to use: Read-heavy workloads (products, reviews, user profiles)

2. Write-Through Cache

Data is written to both the cache and the database simultaneously.

class WriteThroughCache:
    async def set(self, key: str, data: dict, saver):
        """Write to cache and database simultaneously."""
        # Write to database first
        await saver(data)
        # Then update cache
        await self.redis.setex(key, self.ttl, json.dumps(data))
    
    async def delete(self, key: str, deleter):
        """Delete from both cache and database."""
        await deleter()
        await self.redis.delete(key)

When to use: Data that must be consistent (user accounts, orders)

3. Write-Behind (Write-Back) Cache

Data is written to the cache first, then asynchronously flushed to the database.

import asyncio
from collections import OrderedDict

class WriteBehindCache:
    def __init__(self, flush_interval: int = 5):
        self.write_queue = OrderedDict()
        self.flush_interval = flush_interval
    
    async def set(self, key: str, data: dict):
        """Write to cache only. Flush to DB asynchronously."""
        self.write_queue[key] = data
        if len(self.write_queue) >= 100:
            await self._flush()
    
    async def _flush(self):
        """Flush all pending writes to database."""
        batch = dict(self.write_queue)
        self.write_queue.clear()
        # Batch write to database
        await self._batch_save(batch)

When to use: High-write workloads (analytics, metrics, logs)

TTL Strategies

TTL Calculation

def calculate_ttl(data_freshness: str, access_frequency: str) -> int:
    """Calculate optimal TTL based on data characteristics."""
    base_ttl = {
        "real-time": 60,
        "frequent": 300,
        "occasional": 1800,
        "rare": 3600,
    }
    
    frequency_multiplier = {
        "high": 0.5,   # Halve TTL for high-frequency access
        "medium": 1.0,
        "low": 2.0,    # Double TTL for low-frequency access
    }
    
    return base_ttl[data_freshness] * frequency_multiplier[access_frequency]

Cache Invalidation Patterns

1. Time-Based Invalidation

Simplest approach — data expires after TTL.

# Redis automatically handles this with EXPIRE
await redis.setex("user:123", 300, json.dumps(user_data))

2. Event-Based Invalidation

Invalidate cache when data changes.

async def update_user(user_id: str, data: dict):
    await db.update_user(user_id, data)
    # Invalidate cache
    await redis.delete(f"user:{user_id}")
    # Invalidate related caches
    await redis.delete(f"user:{user_id}:profile")
    await redis.delete(f"user:{user_id}:preferences")

3. Tag-Based Invalidation

Group cache entries by tag and invalidate all entries with a tag.

import redis

class TaggedCache:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
    
    async def set(self, key: str, data: dict, tags: list[str], ttl: int = 300):
        """Set cache entry with tags."""
        await self.redis.setex(key, ttl, json.dumps(data))
        for tag in tags:
            await self.redis.sadd(f"tag:{tag}", key)
    
    async def invalidate_tag(self, tag: str):
        """Invalidate all cache entries with a tag."""
        keys = await self.redis.smembers(f"tag:{tag}")
        if keys:
            await self.redis.delete(*keys)
            await self.redis.delete(f"tag:{tag}")

# Usage
cache = TaggedCache(redis_client)
await cache.set("product:123", product_data, tags=["products", "category:electronics"])

# When product catalog updates:
await cache.invalidate_tag("products")

CDN Cache Headers

from fastapi.responses import JSONResponse

@router.get("/products/{product_id}")
async def get_product(product_id: str):
    product = await get_product_from_db(product_id)
    
    response = JSONResponse(content=product)
    # CDN headers
    response.headers["Cache-Control"] = "public, max-age=3600, stale-while-revalidate=300"
    response.headers["ETag"] = f'"{product["updated_at"]}"'
    response.headers["Vary"] = "Accept-Encoding"
    
    return response

Cache-Control Directives

Session Store (Redis)

from fastapi import FastAPI, Request
from redis.asyncio import Redis
import uuid

app = FastAPI()
redis_client = Redis(host="localhost", port=6379, db=0)

@app.middleware("http")
async def session_middleware(request: Request, call_next):
    session_id = request.cookies.get("session_id")
    
    if not session_id:
        session_id = str(uuid.uuid4())
    
    # Load session from Redis
    session_data = await redis_client.get(f"session:{session_id}")
    request.state.session = json.loads(session_data) if session_data else {}
    
    response = await call_next(request)
    
    # Save session to Redis (TTL = 24 hours)
    await redis_client.setex(
        f"session:{session_id}",
        86400,
        json.dumps(request.state.session)
    )
    response.set_cookie("session_id", session_id, httponly=True, secure=True)
    
    return response

When to Cache

Integration with Demeter and Hermes

Demeter uses this skill during schema design to decide what needs caching. Hermes uses this skill during implementation to apply cache patterns.