caching

name	caching
description	Caching strategies and best practices
license	MIT
compatibility	opencode
metadata	{"audience":"developers","category":"performance"}

What I do

• Implement caching at multiple levels
• Choose appropriate cache strategies
• Handle cache invalidation
• Prevent cache stampede
• Use distributed caching effectively
• Monitor cache performance
• Implement cache warming

When to use me

When implementing caching solutions or optimizing cache performance.

Multi-Level Caching

┌─────────────────────────────────────────────────────────────┐
│                     Request                                │
└────────────────────────┬────────────────────────────────────┘
                         │
┌────────────────────────▼────────────────────────────────────┐
│                Browser Cache (HTTP)                         │
│  - Cache-Control headers                                    │
│  - ETag/Last-Modified                                      │
│  - LocalStorage/SessionStorage                              │
└────────────────────────┬────────────────────────────────────┘
                         │
┌────────────────────────▼────────────────────────────────────┐
│                CDN / Edge Cache                              │
│  - CloudFront, Cloudflare                                   │
│  - Static assets, API responses                             │
└────────────────────────┬────────────────────────────────────┘
                         │
┌────────────────────────▼────────────────────────────────────┐
│                Application Cache (Memory)                     │
│  - In-memory caches (LRU, LFU)                              │
│  - Per-process or shared memory                             │
└────────────────────────┬────────────────────────────────────┘
                         │
┌────────────────────────▼────────────────────────────────────┐
│                Distributed Cache (Redis/Memcached)            │
│  - Shared across instances                                  │
│  - High availability                                        │
└────────────────────────┬────────────────────────────────────┘
                         │
┌────────────────────────▼────────────────────────────────────┐
│                Database Cache                                │
│  - Query cache                                             │
│  - Buffer pool                                             │
└─────────────────────────────────────────────────────────────┘

Cache Strategies

import redis
import json
from typing import Optional, Callable, Any
from functools import wraps
from datetime import timedelta


class CacheService:
    """Multi-level cache service with TTL support."""
    
    def __init__(self, redis_url: str) -> None:
        self.redis = redis.from_url(redis_url)
        self.local_cache = {}
    
    def get(self, key: str) -> Optional[Any]:
        """Get from cache hierarchy."""
        # Check local cache first
        if key in self.local_cache:
            value, expires = self.local_cache[key]
            if expires > datetime.utcnow():
                return value
            del self.local_cache[key]
        
        # Check distributed cache
        cached = self.redis.get(key)
        if cached:
            value = json.loads(cached)
            # Populate local cache
            self.local_cache[key] = (
                value,
                datetime.utcnow() + timedelta(seconds=60)
            )
            return value
        
         
    
     () -> :
        
        
        .redis.setex(key, ttl_seconds, json.dumps(value))
        
        
        .local_cache[key] = (
            value,
            datetime.utcnow() + timedelta(seconds=local_ttl_seconds)
        )
    
     () -> :
        
        .redis.delete(key)
        .local_cache.pop(key, )
    
     () -> :
        
        keys = .redis.keys(pattern)
         keys:
            .redis.delete(*keys)
        
         key  (.local_cache.keys()):
             pattern.replace(, )  key:
                 .local_cache[key]


 ():
    
     () -> :

         () -> :
            cache = get_cache_service()
            key = 
            
            cached_value = cache.get(key)
             cached_value   :
                 cached_value
            
            result = func(*args, **kwargs)
            cache.(key, result, ttl)
             result
         wrapper
     decorator

Cache Stampede Prevention

import asyncio
import async_timeout


class CacheStampedePreventer:
    """Prevent cache stampede with distributed locking."""
    
    def __init__(
        self,
        cache: CacheService,
        lock_ttl: int = 30
    ) -> None:
        self.cache = cache
        self.lock_ttl = lock_ttl
    
    async def get_or_compute(
        self,
        key: str,
        compute_fn: Callable,
        ttl: int = 3600
    ) -> Any:
        """Get from cache or compute with stampede prevention."""
        # Try cache first
        cached = self.cache.get(key)
        if cached is not None:
            return cached
        
        # Acquire distributed lock
        lock_key = f"lock:{key}"
        lock_acquired = self.cache.redis.set(
            lock_key,
            "locked",
            nx=True,
            ex=self.lock_ttl
        )
        
        if not lock_acquired:
            # Wait for other process to compute
            return await self._wait_for_computation(key, compute_fn)
        
        :
            
            result =  compute_fn()
            .cache.(key, result, ttl)
             result
        :
            
            .cache.redis.delete(lock_key)
    
      () -> :
        
        max_wait =   
        
         _  (max_wait * ):  
            cached = .cache.get(key)
             cached   :
                 cached
             asyncio.sleep()
        
        
          compute_fn()


 :
    
    
     () -> :
        .cache = cache
        .delta_percent = delta_percent
    
     () -> :
        
        cached = .cache.get(key)
         cached   :
             cached
        
        ttl = .cache.redis.ttl(key)
         ttl == -:  
             compute_fn()
        
        
        threshold = ttl * ( - .delta_percent) / 
         ttl < threshold:
            
            
        
         compute_fn()

HTTP Caching

from fastapi import Request, Response
from typing import Optional


class CacheControlMiddleware:
    """Add appropriate cache control headers."""
    
    def __init__(self, app):
        self.app = app
    
    async def __call__(self, scope, receive, send):
        if scope['type'] != 'http':
            await self.app(scope, receive, send)
            return
        
        async def send_wrapper(message):
            if message['type'] == 'http.response.start':
                request = Request(scope)
                response = Response(
                    content=b'',
                    status_code=message['status'],
                    headers=dict(message.get('headers', []))
                )
                
                # Set cache headers based on route
                if '/api/' in request.url.path:
                    # API endpoints - no cache or short cache
                    if '/static/' not in request.url.path:
                        response.headers['Cache-Control'] = 'no-cache'
                :
                    
                    response.headers[] = 
                
                message[] = (response.headers.items())
            
             send(message)
        
         .app(scope, receive, send_wrapper)

Cache Invalidation Patterns

# Event-driven cache invalidation
class CacheInvalidator:
    def __init__(self, cache: CacheService) -> None:
        self.cache = cache
        self.subscriptions = {}
    
    def subscribe(self, event_type: str, callback: Callable) -> None:
        """Subscribe to invalidation events."""
        if event_type not in self.subscriptions:
            self.subscriptions[event_type] = []
        self.subscriptions[event_type].append(callback)
    
    def invalidate(self, event_type: str, data: dict) -> None:
        """Trigger invalidation based on event."""
        if event_type in self.subscriptions:
            for callback in self.subscriptions[event_type]:
                pattern = callback(data)
                if pattern:
                    self.cache.invalidate_pattern(pattern)
    
    def on_user_update(self, user_id: str) -> None:
        """Invalidate user-related caches."""
        patterns = [
            f"user::*",
            ,
            ,
        ]
         pattern  patterns:
            .cache.invalidate_pattern(pattern)

Cache Monitoring

class CacheMonitor:
    """Monitor cache performance."""
    
    def __init__(self, redis_client: redis.Redis) -> None:
        self.redis = redis_client
    
    def get_stats(self) -> dict:
        """Get cache statistics."""
        info = self.redis.info('stats')
        memory = self.redis.info('memory')
        
        return {
            'hits': info.get('keyspace_hits', 0),
            'misses': info.get('keyspace_misses', 0),
            'hit_rate': self._calculate_hit_rate(info),
            'used_memory': memory.get('used_memory_human'),
            'connected_clients': self.redis.info('clients').get('connected_clients', 0),
            'uptime_seconds': self.redis.info('server').get('uptime_in_seconds', 0),
        }
    
    def _calculate_hit_rate(self, info: dict) -> float:
        hits = info.get('keyspace_hits', 0)
        misses = info.get('keyspace_misses', )
        total = hits + misses
         total == :
             
         (hits / total * , )