Caching

Write-Through Cache

class WriteThrough:
    """Writes go to cache and database simultaneously."""

    def __init__(self, cache_client, database, default_ttl: int = 3600):
        self.cache = cache_client
        self.database = database
        self.default_ttl = default_ttl

    async def write(self, key: str, value: any, ttl: Optional[int] = None):
        # Write to database first
        await self.database.save(key, value)

        # Then update cache
        await self.cache.setex(
            key,
            ttl or self.default_ttl,
            json.dumps(value)
        )

    async def read(self, key: str) -> Optional[any]:
        cached = await self.cache.get(key)
        if cached:
            return json.loads(cached)

        # Fallback to database
        value = await self.database.get(key)
        if value:
            await self.cache.setex(key, self.default_ttl, json.dumps(value))
        return value

Write-Behind (Write-Back) Cache

import asyncio
from collections import defaultdict
from dataclasses import dataclass
from typing import Dict, Any
import time

@dataclass
class PendingWrite:
    key: str
    value: Any
    timestamp: float

class WriteBehind:
    """Writes go to cache immediately, database asynchronously."""

    def __init__(
        self,
        cache_client,
        database,
        flush_interval: float = 5.0,
        batch_size: int = 100
    ):
        self.cache = cache_client
        self.database = database
        self.flush_interval = flush_interval
        self.batch_size = batch_size
        self.pending_writes: Dict[str, PendingWrite] = {}
        self._lock = asyncio.Lock()
        self._flush_task = None

    async def start(self):
        self._flush_task = asyncio.create_task(self._flush_loop())

    async def stop(self):
        if self._flush_task:
            self._flush_task.cancel()
            await self._flush_pending()

    async def write(self, key: str, value: Any):
        # Write to cache immediately
        await self.cache.set(key, json.dumps(value))

        # Queue for database write
        async with self._lock:
            self.pending_writes[key] = PendingWrite(
                key=key,
                value=value,
                timestamp=time.time()
            )

        if len(self.pending_writes) >= self.batch_size:
            await self._flush_pending()

    async def _flush_loop(self):
        while True:
            await asyncio.sleep(self.flush_interval)
            await self._flush_pending()

    async def _flush_pending(self):
        async with self._lock:
            if not self.pending_writes:
                return

            writes = list(self.pending_writes.values())
            self.pending_writes.clear()

        # Batch write to database
        await self.database.batch_save(
            [(w.key, w.value) for w in writes]
        )

Read-Through Cache

class ReadThrough:
    """Cache automatically loads from database on miss."""

    def __init__(self, cache_client, loader, default_ttl: int = 3600):
        self.cache = cache_client
        self.loader = loader
        self.default_ttl = default_ttl

    async def get(self, key: str) -> Optional[Any]:
        # Check cache
        cached = await self.cache.get(key)
        if cached:
            return json.loads(cached)

        # Auto-load on miss
        value = await self.loader(key)
        if value is not None:
            await self.cache.setex(key, self.default_ttl, json.dumps(value))

        return value

2. Invalidation Strategies

from enum import Enum
from typing import Set, List
import fnmatch

class InvalidationStrategy(Enum):
    TIME_BASED = "time_based"
    EVENT_BASED = "event_based"
    VERSION_BASED = "version_based"

class CacheInvalidator:
    def __init__(self, cache_client):
        self.cache = cache_client
        self._tag_index: Dict[str, Set[str]] = defaultdict(set)

    # Tag-based invalidation
    async def set_with_tags(
        self,
        key: str,
        value: Any,
        tags: List[str],
        ttl: int = 3600
    ):
        await self.cache.setex(key, ttl, json.dumps(value))

        for tag in tags:
            self._tag_index[tag].add(key)
            await self.cache.sadd(f"tag:{tag}", key)

    async def invalidate_by_tag(self, tag: str):
        keys = await self.cache.smembers(f"tag:{tag}")
        if keys:
            await self.cache.delete(*keys)
            await self.cache.delete(f"tag:{tag}")

    # Pattern-based invalidation
    async def invalidate_by_pattern(self, pattern: str):
        cursor = 0
        while True:
            cursor, keys = await self.cache.scan(
                cursor,
                match=pattern,
                count=100
            )
            if keys:
                await self.cache.delete(*keys)
            if cursor == 0:
                break

    # Version-based invalidation
    async def get_versioned(self, key: str, version: int) -> Optional[Any]:
        versioned_key = f"{key}:v{version}"
        return await self.cache.get(versioned_key)

    async def set_versioned(
        self,
        key: str,
        value: Any,
        version: int,
        ttl: int = 3600
    ):
        versioned_key = f"{key}:v{version}"
        await self.cache.setex(versioned_key, ttl, json.dumps(value))

# Event-based invalidation with pub/sub
class EventBasedInvalidator:
    def __init__(self, cache_client, pubsub_client):
        self.cache = cache_client
        self.pubsub = pubsub_client

    async def start_listener(self):
        await self.pubsub.subscribe("cache:invalidate")

        async for message in self.pubsub.listen():
            if message["type"] == "message":
                data = json.loads(message["data"])
                await self._handle_invalidation(data)

    async def _handle_invalidation(self, data: dict):
        if "key" in data:
            await self.cache.delete(data["key"])
        elif "pattern" in data:
            await self.invalidate_by_pattern(data["pattern"])
        elif "tag" in data:
            await self.invalidate_by_tag(data["tag"])

    async def publish_invalidation(self, **kwargs):
        await self.pubsub.publish(
            "cache:invalidate",
            json.dumps(kwargs)
        )

3. TTL and Expiration

import random
from datetime import datetime, timedelta

class TTLManager:
    def __init__(self, base_ttl: int = 3600):
        self.base_ttl = base_ttl

    def get_ttl_with_jitter(self, ttl: Optional[int] = None) -> int:
        """Add randomness to prevent synchronized expiration."""
        base = ttl or self.base_ttl
        jitter = random.uniform(-0.1, 0.1) * base
        return int(base + jitter)

    def get_sliding_ttl(
        self,
        last_access: datetime,
        min_ttl: int = 60,
        max_ttl: int = 3600
    ) -> int:
        """TTL based on access frequency."""
        age = (datetime.utcnow() - last_access).total_seconds()

        if age < 60:
            return max_ttl  # Frequently accessed
        elif age < 300:
            return max_ttl // 2
        else:
            return min_ttl

    def get_tiered_ttl(self, data_type: str) -> int:
        """Different TTL for different data types."""
        ttl_tiers = {
            "user_session": 86400,      # 1 day
            "user_profile": 3600,       # 1 hour
            "product_catalog": 300,     # 5 minutes
            "search_results": 60,       # 1 minute
            "real_time_data": 10,       # 10 seconds
        }
        return ttl_tiers.get(data_type, self.base_ttl)

# Sliding window expiration
class SlidingCache:
    def __init__(self, cache_client, default_ttl: int = 3600):
        self.cache = cache_client
        self.default_ttl = default_ttl

    async def get(self, key: str) -> Optional[Any]:
        value = await self.cache.get(key)
        if value:
            # Refresh TTL on access
            await self.cache.expire(key, self.default_ttl)
            return json.loads(value)
        return None

    async def set(self, key: str, value: Any, ttl: Optional[int] = None):
        await self.cache.setex(
            key,
            ttl or self.default_ttl,
            json.dumps(value)
        )

4. Redis/Memcached Patterns

import redis.asyncio as redis
from typing import List, Tuple

class RedisCache:
    def __init__(self, redis_url: str):
        self.redis = redis.from_url(redis_url)

    # Hash operations for structured data
    async def set_hash(self, key: str, data: dict, ttl: int = 3600):
        await self.redis.hset(key, mapping=data)
        await self.redis.expire(key, ttl)

    async def get_hash(self, key: str) -> Optional[dict]:
        data = await self.redis.hgetall(key)
        return {k.decode(): v.decode() for k, v in data.items()} if data else None

    async def update_hash_field(self, key: str, field: str, value: str):
        await self.redis.hset(key, field, value)

    # Sorted sets for leaderboards/rankings
    async def add_to_ranking(self, key: str, member: str, score: float):
        await self.redis.zadd(key, {member: score})

    async def get_top_n(self, key: str, n: int) -> List[Tuple[str, float]]:
        return await self.redis.zrevrange(key, 0, n - 1, withscores=True)

    # Lists for queues
    async def push_to_queue(self, key: str, *values):
        await self.redis.lpush(key, *values)

    async def pop_from_queue(self, key: str, timeout: int = 0):
        return await self.redis.brpop(key, timeout=timeout)

    # Pub/Sub for cache invalidation
    async def publish(self, channel: str, message: str):
        await self.redis.publish(channel, message)

    # Lua scripts for atomic operations
    async def increment_with_cap(self, key: str, cap: int) -> int:
        script = """
        local current = redis.call('GET', KEYS[1])
        if current and tonumber(current) >= tonumber(ARGV[1]) then
            return -1
        end
        return redis.call('INCR', KEYS[1])
        """
        return await self.redis.eval(script, 1, key, cap)

    # Pipeline for batch operations
    async def batch_get(self, keys: List[str]) -> List[Optional[str]]:
        async with self.redis.pipeline() as pipe:
            for key in keys:
                pipe.get(key)
            return await pipe.execute()

    async def batch_set(
        self,
        items: List[Tuple[str, str]],
        ttl: int = 3600
    ):
        async with self.redis.pipeline() as pipe:
            for key, value in items:
                pipe.setex(key, ttl, value)
            await pipe.execute()

5. Cache Stampede Prevention

import asyncio
import hashlib
import time
from typing import Optional, Callable

class StampedeProtection:
    """Prevents cache stampede using various strategies."""

    def __init__(self, cache_client):
        self.cache = cache_client
        self._locks: Dict[str, asyncio.Lock] = {}

    # Strategy 1: Locking (prevents concurrent regeneration)
    async def get_with_lock(
        self,
        key: str,
        loader: Callable,
        ttl: int = 3600
    ) -> Any:
        cached = await self.cache.get(key)
        if cached:
            return json.loads(cached)

        # Get or create lock for this key
        if key not in self._locks:
            self._locks[key] = asyncio.Lock()

        async with self._locks[key]:
            # Double-check after acquiring lock
            cached = await self.cache.get(key)
            if cached:
                return json.loads(cached)

            value = await loader()
            await self.cache.setex(key, ttl, json.dumps(value))
            return value

    # Strategy 2: Probabilistic early expiration
    async def get_with_early_recompute(
        self,
        key: str,
        loader: Callable,
        ttl: int = 3600,
        beta: float = 1.0
    ) -> Any:
        data = await self.cache.get(key)

        if data:
            cached = json.loads(data)
            expiry = cached.get("_expiry", 0)
            delta = cached.get("_delta", 0)

            # Probabilistic early recomputation
            now = time.time()
            if now - delta * beta * random.random() < expiry:
                return cached["value"]

        # Recompute
        start = time.time()
        value = await loader()
        delta = time.time() - start

        cache_data = {
            "value": value,
            "_expiry": time.time() + ttl,
            "_delta": delta
        }
        await self.cache.setex(key, ttl, json.dumps(cache_data))
        return value

    # Strategy 3: Stale-while-revalidate
    async def get_stale_while_revalidate(
        self,
        key: str,
        loader: Callable,
        ttl: int = 3600,
        stale_ttl: int = 300
    ) -> Any:
        data = await self.cache.get(key)

        if data:
            cached = json.loads(data)

            if cached.get("_fresh", True):
                return cached["value"]

            # Return stale, revalidate in background
            asyncio.create_task(self._revalidate(key, loader, ttl, stale_ttl))
            return cached["value"]

        return await self._load_and_cache(key, loader, ttl, stale_ttl)

    async def _revalidate(
        self,
        key: str,
        loader: Callable,
        ttl: int,
        stale_ttl: int
    ):
        lock_key = f"lock:{key}"
        acquired = await self.cache.setnx(lock_key, "1")

        if acquired:
            try:
                await self.cache.expire(lock_key, 30)
                await self._load_and_cache(key, loader, ttl, stale_ttl)
            finally:
                await self.cache.delete(lock_key)

    async def _load_and_cache(
        self,
        key: str,
        loader: Callable,
        ttl: int,
        stale_ttl: int
    ):
        value = await loader()

        cache_data = {"value": value, "_fresh": True}
        await self.cache.setex(key, ttl, json.dumps(cache_data))

        # Mark as stale after TTL
        async def mark_stale():
            await asyncio.sleep(ttl - stale_ttl)
            data = await self.cache.get(key)
            if data:
                cached = json.loads(data)
                cached["_fresh"] = False
                await self.cache.setex(key, stale_ttl, json.dumps(cached))

        asyncio.create_task(mark_stale())
        return value

6. Distributed Caching

import hashlib
from typing import List, Optional
import random

class ConsistentHashing:
    """Consistent hashing for distributed cache nodes."""

    def __init__(self, nodes: List[str], replicas: int = 100):
        self.replicas = replicas
        self.ring: Dict[int, str] = {}
        self.sorted_keys: List[int] = []

        for node in nodes:
            self.add_node(node)

    def _hash(self, key: str) -> int:
        return int(hashlib.md5(key.encode()).hexdigest(), 16)

    def add_node(self, node: str):
        for i in range(self.replicas):
            key = self._hash(f"{node}:{i}")
            self.ring[key] = node
            self.sorted_keys.append(key)
        self.sorted_keys.sort()

    def remove_node(self, node: str):
        for i in range(self.replicas):
            key = self._hash(f"{node}:{i}")
            del self.ring[key]
            self.sorted_keys.remove(key)

    def get_node(self, key: str) -> str:
        if not self.ring:
            raise ValueError("No nodes in ring")

        h = self._hash(key)

        for ring_key in self.sorted_keys:
            if h <= ring_key:
                return self.ring[ring_key]

        return self.ring[self.sorted_keys[0]]

class DistributedCache:
    """Distributed cache with consistent hashing."""

    def __init__(self, nodes: List[str]):
        self.ring = ConsistentHashing(nodes)
        self.clients: Dict[str, RedisCache] = {
            node: RedisCache(node) for node in nodes
        }

    def _get_client(self, key: str) -> RedisCache:
        node = self.ring.get_node(key)
        return self.clients[node]

    async def get(self, key: str) -> Optional[str]:
        client = self._get_client(key)
        return await client.redis.get(key)

    async def set(self, key: str, value: str, ttl: int = 3600):
        client = self._get_client(key)
        await client.redis.setex(key, ttl, value)

    async def delete(self, key: str):
        client = self._get_client(key)
        await client.redis.delete(key)

    # Multi-get across nodes
    async def mget(self, keys: List[str]) -> Dict[str, Optional[str]]:
        # Group keys by node
        node_keys: Dict[str, List[str]] = defaultdict(list)
        for key in keys:
            node = self.ring.get_node(key)
            node_keys[node].append(key)

        # Fetch in parallel
        results = {}
        tasks = []

        for node, node_key_list in node_keys.items():
            client = self.clients[node]
            tasks.append(self._fetch_from_node(client, node_key_list, results))

        await asyncio.gather(*tasks)
        return results

    async def _fetch_from_node(
        self,
        client: RedisCache,
        keys: List[str],
        results: Dict
    ):
        values = await client.batch_get(keys)
        for key, value in zip(keys, values):
            results[key] = value

7. Database Query Caching

import sqlalchemy
from typing import Optional, List, Any
import hashlib

class QueryCache:
    """Database query result caching with automatic invalidation."""

    def __init__(self, cache_client, default_ttl: int = 300):
        self.cache = cache_client
        self.default_ttl = default_ttl

    def _query_key(self, sql: str, params: tuple) -> str:
        """Generate cache key from SQL and parameters."""
        query_str = f"{sql}:{params}"
        return f"query:{hashlib.md5(query_str.encode()).hexdigest()}"

    async def execute_cached(
        self,
        session,
        query,
        params: Optional[dict] = None,
        ttl: Optional[int] = None
    ) -> List[Any]:
        sql_str = str(query)
        cache_key = self._query_key(sql_str, tuple(sorted((params or {}).items())))

        # Check cache
        cached = await self.cache.get(cache_key)
        if cached:
            return json.loads(cached)

        # Execute query
        result = session.execute(query, params).fetchall()
        serialized = [dict(row) for row in result]

        # Cache result
        await self.cache.setex(
            cache_key,
            ttl or self.default_ttl,
            json.dumps(serialized)
        )

        return serialized

    async def invalidate_table(self, table_name: str):
        """Invalidate all queries for a table."""
        pattern = f"query:*{table_name}*"
        await self.cache.delete_pattern(pattern)

# ORM-level caching with SQLAlchemy
from sqlalchemy import event
from sqlalchemy.orm import Session

class ORMCache:
    def __init__(self, cache_client):
        self.cache = cache_client

    def setup_listeners(self, engine):
        """Set up automatic cache invalidation on writes."""

        @event.listens_for(Session, "after_flush")
        def receive_after_flush(session, flush_context):
            # Invalidate cache for modified tables
            for obj in session.dirty | session.new | session.deleted:
                table = obj.__tablename__
                asyncio.create_task(
                    self.cache.delete_pattern(f"query:*{table}*")
                )

    async def get_by_id(self, model, obj_id: int, session):
        key = f"{model.__tablename__}:{obj_id}"
        cached = await self.cache.get(key)

        if cached:
            return json.loads(cached)

        obj = session.query(model).get(obj_id)
        if obj:
            await self.cache.setex(key, 3600, json.dumps(obj.to_dict()))

        return obj

8. CDN and Static Asset Caching

from datetime import datetime, timedelta
from typing import Dict, Optional

class CDNCache:
    """CDN caching strategies with proper headers."""

    @staticmethod
    def get_cache_headers(
        cache_type: str,
        max_age: int = 3600
    ) -> Dict[str, str]:
        """Generate appropriate HTTP cache headers."""

        strategies = {
            "immutable": {
                "Cache-Control": f"public, max-age={max_age}, immutable",
                "Expires": (datetime.utcnow() + timedelta(seconds=max_age)).strftime(
                    "%a, %d %b %Y %H:%M:%S GMT"
                ),
            },
            "versioned": {
                "Cache-Control": f"public, max-age={max_age}",
                "ETag": None,  # Set dynamically
            },
            "revalidate": {
                "Cache-Control": "public, max-age=0, must-revalidate",
                "ETag": None,  # Set dynamically
            },
            "private": {
                "Cache-Control": "private, max-age=0, must-revalidate",
                "Pragma": "no-cache",
            },
            "no-cache": {
                "Cache-Control": "no-store, no-cache, must-revalidate, max-age=0",
                "Pragma": "no-cache",
                "Expires": "0",
            },
        }

        return strategies.get(cache_type, strategies["revalidate"])

    @staticmethod
    def generate_etag(content: bytes) -> str:
        """Generate ETag from content hash."""
        return hashlib.md5(content).hexdigest()

# FastAPI/Flask example
from fastapi import Response
from fastapi.responses import FileResponse

class StaticAssetCache:
    def __init__(self, cdn_base_url: Optional[str] = None):
        self.cdn_base_url = cdn_base_url

    def serve_asset(
        self,
        file_path: str,
        asset_type: str = "immutable"
    ) -> FileResponse:
        """Serve static asset with caching headers."""

        with open(file_path, "rb") as f:
            content = f.read()

        headers = CDNCache.get_cache_headers(
            asset_type,
            max_age=31536000 if asset_type == "immutable" else 3600
        )

        # Add ETag for versioned assets
        if asset_type in ["versioned", "revalidate"]:
            headers["ETag"] = f'"{CDNCache.generate_etag(content)}"'

        return FileResponse(
            file_path,
            headers=headers,
            media_type=self._get_media_type(file_path)
        )

    def get_asset_url(self, path: str, version: Optional[str] = None) -> str:
        """Generate CDN URL with optional versioning."""
        base = self.cdn_base_url or ""

        if version:
            # Append version to filename for cache busting
            parts = path.rsplit(".", 1)
            path = f"{parts[0]}.{version}.{parts[1]}"

        return f"{base}/{path}"

    @staticmethod
    def _get_media_type(file_path: str) -> str:
        ext_to_mime = {
            ".js": "application/javascript",
            ".css": "text/css",
            ".png": "image/png",
            ".jpg": "image/jpeg",
            ".svg": "image/svg+xml",
            ".woff2": "font/woff2",
        }
        ext = file_path[file_path.rfind("."):]
        return ext_to_mime.get(ext, "application/octet-stream")

9. ML Model Caching

import pickle
from typing import Any, Callable
import numpy as np

class MLModelCache:
    """Caching for machine learning models and predictions."""

    def __init__(self, cache_client, model_store_path: str = "./models"):
        self.cache = cache_client
        self.model_store = model_store_path

    async def get_model(self, model_id: str, version: str):
        """Load model from cache or disk."""
        key = f"model:{model_id}:v{version}"

        cached = await self.cache.get(key)
        if cached:
            return pickle.loads(cached)

        # Load from disk
        path = f"{self.model_store}/{model_id}/{version}/model.pkl"
        with open(path, "rb") as f:
            model = pickle.load(f)

        # Cache serialized model (use compression for large models)
        await self.cache.setex(
            key,
            86400,  # 1 day
            pickle.dumps(model)
        )

        return model

    async def cache_prediction(
        self,
        model_id: str,
        input_hash: str,
        prediction: Any,
        ttl: int = 3600
    ):
        """Cache prediction results."""
        key = f"pred:{model_id}:{input_hash}"
        await self.cache.setex(key, ttl, json.dumps(prediction))

    async def get_cached_prediction(
        self,
        model_id: str,
        input_data: Any
    ) -> Optional[Any]:
        """Retrieve cached prediction if available."""
        input_hash = hashlib.md5(
            json.dumps(input_data, sort_keys=True).encode()
        ).hexdigest()

        key = f"pred:{model_id}:{input_hash}"
        cached = await self.cache.get(key)

        return json.loads(cached) if cached else None

# Feature caching for ML pipelines
class FeatureCache:
    def __init__(self, cache_client):
        self.cache = cache_client

    async def get_features(
        self,
        entity_id: str,
        feature_names: List[str],
        compute_fn: Callable
    ) -> Dict[str, Any]:
        """Get features from cache or compute."""

        # Try to get from cache
        keys = [f"feature:{entity_id}:{name}" for name in feature_names]
        cached_values = await self.cache.mget(keys)

        features = {}
        missing = []

        for name, value in zip(feature_names, cached_values):
            if value:
                features[name] = json.loads(value)
            else:
                missing.append(name)

        # Compute missing features
        if missing:
            computed = await compute_fn(entity_id, missing)

            # Cache computed features
            for name, value in computed.items():
                key = f"feature:{entity_id}:{name}"
                await self.cache.setex(key, 3600, json.dumps(value))
                features[name] = value

        return features

# Embeddings cache for vector similarity
class EmbeddingCache:
    def __init__(self, cache_client):
        self.cache = cache_client

    async def get_embedding(
        self,
        text: str,
        model: str,
        embed_fn: Callable
    ) -> np.ndarray:
        """Get or compute text embedding."""

        text_hash = hashlib.md5(text.encode()).hexdigest()
        key = f"embed:{model}:{text_hash}"

        cached = await self.cache.get(key)
        if cached:
            return np.frombuffer(cached, dtype=np.float32)

        embedding = await embed_fn(text)

        # Store as binary
        await self.cache.setex(
            key,
            86400 * 7,  # 1 week
            embedding.tobytes()
        )

        return embedding

10. Local In-Memory Caching

from collections import OrderedDict
from threading import RLock
from typing import Optional
import time

class LRUCache:
    """Thread-safe LRU cache with TTL support."""

    def __init__(self, capacity: int = 1000, default_ttl: Optional[int] = None):
        self.cache = OrderedDict()
        self.capacity = capacity
        self.default_ttl = default_ttl
        self.lock = RLock()

    def get(self, key: str) -> Optional[Any]:
        with self.lock:
            if key not in self.cache:
                return None

            value, expiry = self.cache[key]

            # Check expiration
            if expiry and time.time() > expiry:
                del self.cache[key]
                return None

            # Move to end (most recently used)
            self.cache.move_to_end(key)
            return value

    def put(self, key: str, value: Any, ttl: Optional[int] = None):
        with self.lock:
            if key in self.cache:
                self.cache.move_to_end(key)

            ttl = ttl or self.default_ttl
            expiry = time.time() + ttl if ttl else None
            self.cache[key] = (value, expiry)

            if len(self.cache) > self.capacity:
                # Remove oldest (least recently used)
                self.cache.popitem(last=False)

    def invalidate(self, key: str):
        with self.lock:
            self.cache.pop(key, None)

    def clear(self):
        with self.lock:
            self.cache.clear()

# Function memoization decorator
from functools import wraps

def memoize(ttl: Optional[int] = None, maxsize: int = 128):
    """Memoization decorator with TTL."""

    cache = LRUCache(capacity=maxsize, default_ttl=ttl)

    def decorator(func: Callable):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # Create cache key from args
            key = f"{func.__name__}:{args}:{sorted(kwargs.items())}"

            result = cache.get(key)
            if result is not None:
                return result

            result = func(*args, **kwargs)
            cache.put(key, result)
            return result

        wrapper.cache_clear = cache.clear
        wrapper.cache_info = lambda: {
            "size": len(cache.cache),
            "maxsize": cache.capacity
        }

        return wrapper

    return decorator

# Usage
@memoize(ttl=300, maxsize=1000)
def expensive_computation(x: int, y: int) -> int:
    return x ** y

Best Practices

1. Cache What Matters: Focus on frequently accessed, expensive-to-compute data.

2. Set Appropriate TTLs: Balance freshness vs performance. Use jitter to prevent thundering herd.

3. Handle Cache Failures Gracefully: Cache should enhance, not be required for operation.

4. Monitor Cache Performance: Track hit rates, latency, and memory usage.

5. Use Appropriate Data Structures: Hashes for objects, sorted sets for rankings, lists for queues.

6. Implement Proper Invalidation: Prefer event-driven invalidation over TTL alone for critical data.

7. Prevent Stampedes: Use locking, early recomputation, or stale-while-revalidate.

8. Size Your Cache Appropriately: Monitor eviction rates and adjust size accordingly.

9. Layer Your Caches: Browser cache → CDN → Redis → Database for optimal performance.

10. Security Considerations: Never cache sensitive data (passwords, tokens, PII) without encryption.

Examples

Complete Caching Layer

class CacheLayer:
    """Production-ready caching layer with multiple strategies."""

    def __init__(self, redis_url: str, default_ttl: int = 3600):
        self.redis = RedisCache(redis_url)
        self.stampede = StampedeProtection(self.redis.redis)
        self.ttl_manager = TTLManager(default_ttl)
        self.invalidator = CacheInvalidator(self.redis.redis)

    async def get_user(self, user_id: str) -> User:
        return await self.stampede.get_with_lock(
            f"user:{user_id}",
            lambda: self.database.get_user(user_id),
            ttl=self.ttl_manager.get_tiered_ttl("user_profile")
        )

    async def update_user(self, user_id: str, data: dict):
        await self.database.update_user(user_id, data)
        await self.invalidator.invalidate_by_tag(f"user:{user_id}")

    async def get_product_listing(self, category: str) -> List[Product]:
        return await self.stampede.get_stale_while_revalidate(
            f"products:{category}",
            lambda: self.database.get_products(category),
            ttl=self.ttl_manager.get_tiered_ttl("product_catalog"),
            stale_ttl=60
        )