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

Strategy	TTL	Use Case
Short-lived	1-5 min	Real-time data (stock prices, live scores)
Medium-lived	5-30 min	User sessions, API responses
Long-lived	1-24 hours	Product catalogs, static content
Never expire	No TTL	Reference data (countries, currencies)

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

Directive	Meaning
public	Cacheable by any cache
private	Cacheable only by browser
max-age=N	Cache for N seconds
no-cache	Must revalidate before using cache
no-store	Do not cache at all
stale-while-revalidate=N	Serve stale for N seconds while revalidating
stale-if-error=N	Serve stale for N seconds if origin errors

---

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

Scenario	Cache?	Pattern
Read-heavy, rarely changing	✅ Yes	Read-through, long TTL
Write-heavy, real-time	❌ No	Direct DB access
Expensive computation	✅ Yes	Read-through, medium TTL
User-specific data	✅ Yes	Read-through, short TTL
Static content	✅ Yes	CDN, long TTL
Financial transactions	❌ No	Direct DB, no 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.