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)