Caching Strategies 2026: Redis, Valkey, CDN, and Application Caching

┌─────────────────────────────────────┐
│        CDN / Edge Cache             │  ← Milliseconds, global
├─────────────────────────────────────┤
│        Application Cache            │  ← Milliseconds
├─────────────────────────────────────┤
│        Database Query Cache         │  ← Milliseconds
├─────────────────────────────────────┤
│        Primary Database             │  ← Milliseconds to seconds
└─────────────────────────────────────┘

import redis
import json

r = redis.Redis(host='localhost', port=6379, db=0)

# Strings - Simple values
r.set('user:123:name', 'John Doe')
r.set('user:123:email', '[email protected]')
r.setex('session:abc123', 3600, json.dumps({'user_id': 123}))

# Hashes - Objects
r.hset('user:123', mapping={
    'name': 'John Doe',
    'email': '[email protected]',
    'created_at': '2024-01-15'
})
user = r.hgetall('user:123')

# Lists - Ordered collections
r.lpush('recent_searches:user123', 'shoes')
r.ltrim('recent_searches:user123', 0, 9)  # Keep only 10

# Sets - Unique collections
r.sadd('product:views:2024-01-15', 'product1', 'product2', 'product3')
unique_viewers = r.scard('product:views:2024-01-15')

# Sorted Sets - Leaderboards
r.zadd('leaderboard:score', {'player1': 100, 'player2': 95, 'player3': 90})
top_players = r.zrevrange('leaderboard:score', 0, 9, withscores=True)

# Standard cache-aside pattern
def get_user(user_id):
    # Try cache first
    cached = redis.get(f'user:{user_id}')
    if cached:
        return json.loads(cached)
    
    # Cache miss - fetch from database
    user = db.query('SELECT * FROM users WHERE id = ?', user_id)
    
    if user:
        # Store in cache with TTL
        redis.setex(f'user:{user_id}', 3600, json.dumps(user))
    
    return user

def update_user(user_id, data):
    # Update database first
    db.execute('UPDATE users SET ... WHERE id = ?', user_id, data)
    
    # Invalidate cache
    redis.delete(f'user:{user_id}')

# Write-through - cache updated on every write
def create_order(order_data):
    # Write to database
    order_id = db.insert('orders', order_data)
    
    # Write to cache immediately
    cache_key = f'order:{order_id}'
    redis.setex(cache_key, 3600, json.dumps({**order_data, 'id': order_id}))
    
    return order_id

def update_order(order_id, data):
    db.update('orders', data, 'id = ?', order_id)
    
    # Update cache
    cache_key = f'order:{order_id}'
    cached = redis.get(cache_key)
    if cached:
        order = json.loads(cached)
        order.update(data)
        redis.setex(cache_key, 3600, json.dumps(order))

# Write-behind - async database writes
import asyncio
from collections import deque

write_queue = deque()

async def update_user_async(user_id, data):
    # Update cache immediately
    redis.setex(f'user:{user_id}', 3600, json.dumps(data))
    
    # Queue for async database write
    write_queue.append({
        'table': 'users',
        'id': user_id,
        'data': data
    })

async def flush_writes():
    while True:
        if write_queue:
            batch = []
            while write_queue and len(batch) < 100:
                batch.append(write_queue.popleft())
            
            # Bulk write to database
            db.batch_update('users', batch)
        
        await asyncio.sleep(1)

# Redis Cluster connection
from redis.cluster import RedisCluster

nodes = [
    {'host': 'redis-1', 'port': 6379},
    {'host': 'redis-2', 'port': 6379},
    {'host': 'redis-3', 'port': 6379},
]

rc = RedisCluster(startup_nodes=nodes, decode_responses=True)

# Automatic key distribution
rc.set('user:1:name', 'Alice')
rc.set('user:2:name', 'Bob')
# Keys automatically sharded across nodes

# Publisher - notify other services of changes
def invalidate_user_cache(user_id):
    # Local cache delete
    redis.delete(f'user:{user_id}')
    
    # Notify other instances
    redis.publish('cache:invalidate', json.dumps({
        'key': f'user:{user_id}',
        'pattern': f'user:{user_id}:*'
    }))

# Subscriber - listen for invalidations
def listen_for_invalidation():
    pubsub = redis.pubsub()
    pubsub.subscribe('cache:invalidate')
    
    for message in pubsub.listen():
        if message['type'] == 'message':
            data = json.loads(message['data'])
            if '*' in data['pattern']:
                # Handle pattern
                for key in redis.keys(data['pattern']):
                    redis.delete(key)
            else:
                redis.delete(data['key'])

# /etc/valkey/valkey.conf
io-threads 4
io-threads-do-reads yes
maxmemory 4gb
maxmemory-policy allkeys-lfu
save 900 1
save 300 10
appendonly yes
appendfsync everysec
tls-port 6379
tls-cert-file /etc/valkey/valkey.crt
tls-key-file /etc/valkey/valkey.key

$ valkey-cli INFO server | grep valkey_version
valkey_version:8.1.7

# On AWS ElastiCache — one-click upgrade
aws elasticache modify-cache-cluster \
  --cache-cluster-id my-redis-cluster \
  --engine valkey \
  --engine-version 8.1

# Self-hosted — export RDB and import
redis-cli --rdb dump.rdb
valkey-server --rdb-compression yes < dump.rdb

from functools import lru_cache
from threading import Lock
import time

# Simple in-memory cache with TTL
class Cache:
    def __init__(self, ttl=300):
        self._cache = {}
        self._timestamps = {}
        self._ttl = ttl
        self._lock = Lock()
    
    def get(self, key):
        with self._lock:
            if key in self._cache:
                if time.time() - self._timestamps[key] < self._ttl:
                    return self._cache[key]
                del self._cache[key]
                del self._timestamps[key]
        return None
    
    def set(self, key, value):
        with self._lock:
            self._cache[key] = value
            self._timestamps[key] = time.time()
    
    def delete(self, key):
        with self._lock:
            self._cache.pop(key, None)
            self._timestamps.pop(key, None)
    
    def clear(self):
        with self._lock:
            self._cache.clear()
            self._timestamps.clear()

# Usage with LRU cache
@lru_cache(maxsize=1000)
def get_product_category(product_id):
    # Expensive database query
    return db.query('SELECT category FROM products WHERE id = ?', product_id)

# Manual cache management
from cachetools import TTLCache, cached

product_cache = TTLCache(maxsize=10000, ttl=300)

@cached(cache=product_cache)
def get_product(product_id):
    return db.query('SELECT * FROM products WHERE id = ?', product_id)

# Multi-layer cache with local + Redis
class MultiLayerCache:
    def __init__(self, local_ttl=60, remote_ttl=3600):
        self.local = {}  # Local dict
        self.redis = redis.Redis()
        self.local_ttl = local_ttl
        self.remote_ttl = remote_ttl
    
    def get(self, key):
        # Check local cache first
        if key in self.local:
            value, timestamp = self.local[key]
            if time.time() - timestamp < self.local_ttl:
                return value
            del self.local[key]
        
        # Check Redis
        value = self.redis.get(key)
        if value:
            # Populate local cache
            self.local[key] = (value, time.time())
            return json.loads(value)
        
        return None
    
    def set(self, key, value):
        # Write to both layers
        self.redis.setex(key, self.remote_ttl, json.dumps(value))
        self.local[key] = (value, time.time())

# Static assets with content hash in filename — cache forever
def get_hashed_asset(filename):
    response = send_file(f'static/{filename}')
    response.headers['Cache-Control'] = 'public, max-age=31536000, immutable'
    return response

# HTML pages — CDN caches briefly, stale-while-revalidate for resilience
def get_page(slug):
    response = render_page(slug)
    response.headers['Cache-Control'] = 'public, max-age=0, s-maxage=300, stale-while-revalidate=86400'
    return response

# API responses — short TTL with background refresh
def get_api_data(request):
    if request.user:
        response = jsonify(fetch_user_data(request.user))
        response.headers['Cache-Control'] = 'private, max-age=60, must-revalidate'
    else:
        response = jsonify(fetch_public_data())
        response.headers['Cache-Control'] = 'public, max-age=60, s-maxage=300, stale-while-revalidate=86400'
    return response

# Sensitive data
def get_user_dashboard(request):
    response = jsonify(request.user.dashboard)
    response.headers['Cache-Control'] = 'private, no-store'
    return response

from hashlib import md5

def get_user_profile(user_id):
    profile = db.query('SELECT * FROM users WHERE id = ?', user_id)
    profile_json = json.dumps(profile)
    etag = md5(profile_json.encode()).hexdigest()
    
    # Client sends If-None-Match header
    if request.headers.get('If-None-Match') == etag:
        return Response(status=304)  # Not Modified — no body
    
    response = jsonify(profile)
    response.headers['ETag'] = etag
    response.headers['Cache-Control'] = 'public, max-age=3600'
    return response

response.headers['Vary'] = 'Accept-Encoding, Accept-Language'

import boto3
import time

# CloudFront invalidation
def invalidate_cdn(paths):
    cloudfront = boto3.client('cloudfront')
    response = cloudfront.create_invalidation(
        DistributionId='E1234567890ABC',
        InvalidationBatch={
            'CallerReference': f'invalidation-{int(time.time())}',
            'Paths': {
                'Quantity': len(paths),
                'Items': paths
            }
        }
    )
    return response['Invalidation']['Id']

invalidate_cdn(['/api/products/*', '/static/images/*'])

// Cloudflare cache purge by URL or tag
async function purgeCache(zoneId, apiToken, urls) {
  const response = await fetch(
    `https://api.cloudflare.com/client/v4/zones/${zoneId}/purge_cache`,
    {
      method: 'POST',
      headers: {
        'Authorization': `Bearer ${apiToken}`,
        'Content-Type': 'application/json'
      },
      body: JSON.stringify({
        files: urls,
        tags: ['product-update'],
        prefixes: ['/api/']
      })
    }
  );
  return response.json();
}

HTTP/1.1 200 OK
Cache-Tag: user-123, post-456, blog-list

curl -X POST https://api.fastly.com/service/{id}/purge/user-123

Traditional caching:
  [cache expires] → [origin request] → [wait] → [response]
                   ↑ user waits

Stale-While-Revalidate:
  [cache expires] → [serve stale instantly] → [background refresh]
                   ↑ user gets instant response

User 1 at 60s: [cache hit, fresh] — instant
User 2 at 61s: [cache hit, stale] — instant, background refresh starts
User 3 at 62s: [cache hit, fresh] — instant (just refreshed from bg)

Cache-Control: max-age=60, stale-while-revalidate=86400, stale-if-error=86400

# Fixed TTL
CACHE_TTL = {
    'user_profile': 3600,      # 1 hour
    'product_list': 300,        # 5 minutes
    'config': 86400,           # 24 hours
    'analytics': 60,            # 1 minute
}

def get_cached(key, ttl_key, fetcher):
    cached = redis.get(key)
    if cached:
        return json.loads(cached)
    
    value = fetcher()
    redis.setex(key, CACHE_TTL[ttl_key], json.dumps(value))
    return value

# Invalidate on data changes
def on_product_updated(product_id):
    # Invalidate specific cache
    redis.delete(f'product:{product_id}')
    
    # Invalidate list caches
    for pattern in ['products:list:*', 'products:category:*']:
        for key in redis.keys(pattern):
            redis.delete(key)
    
    # Notify other services
    redis.publish('cache:invalidate', {
        'type': 'product',
        'id': product_id
    })

# Serve stale content while refreshing in background
def get_product_with_revalidation(product_id):
    cache_key = f'product:{product_id}'
    
    # Try to get from cache
    cached = redis.get(cache_key)
    
    if cached:
        product = json.loads(cached)
        
        # Check if stale
        is_stale = redis.ttl(cache_key) < 60
        
        if is_stale:
            # Schedule background refresh (don't block)
            asyncio.create_task(refresh_product_cache(product_id))
        
        return product
    
    # Cache miss - fetch synchronously
    return refresh_product_cache(product_id)

async def refresh_product_cache(product_id):
    product = db.get_product(product_id)
    
    # Update cache
    redis.setex(f'product:{product_id}', 3600, json.dumps(product))
    
    return product

import random
import math

def should_refresh_early(ttl_remaining, total_ttl):
    if ttl_remaining <= 0:
        return True
    # Probability increases as TTL approaches zero
    prob = math.exp(-ttl_remaining / (total_ttl * 0.5))
    return random.random() < prob

def get_with_probabilistic_refresh(key, ttl=3600, fetch_fn=None):
    cached = redis.get(key)
    if cached:
        ttl_remaining = redis.ttl(key)
        if should_refresh_early(ttl_remaining, ttl):
            # Only some requests refresh, preventing stampede
            asyncio.create_task(refresh_async(key, ttl, fetch_fn))
        return json.loads(cached)
    
    value = fetch_fn()
    redis.setex(key, ttl, json.dumps(value))
    return value

def get_with_lock(key, fetch_fn, ttl=3600):
    cached = redis.get(key)
    if cached:
        return json.loads(cached)
    
    # Acquire a distributed lock — only one request proceeds
    lock_key = f'lock:{key}'
    lock = redis.lock(lock_key, timeout=10, blocking_timeout=5)
    
    if lock.acquire():
        try:
            # Double-check — another request may have populated it
            cached = redis.get(key)
            if cached:
                return json.loads(cached)
            
            value = fetch_fn()
            redis.setex(key, ttl, json.dumps(value))
            return value
        finally:
            lock.release()
    else:
        # Another request is fetching — wait briefly, then retry
        time.sleep(0.05)
        return get_with_lock(key, fetch_fn, ttl)

from collections import defaultdict
import asyncio

class SingleFlightCache:
    def __init__(self):
        self._in_flight = defaultdict(list)
    
    async def get(self, key, fetch_fn, ttl=3600):
        cached = redis.get(key)
        if cached:
            return json.loads(cached)
        
        # If another request is already fetching, wait for it
        if key in self._in_flight:
            future = asyncio.Future()
            self._in_flight[key].append(future)
            return await future
        
        # We are the designated fetcher
        self._in_flight[key] = []
        try:
            value = fetch_fn()
            redis.setex(key, ttl, json.dumps(value))
            # Notify waiters
            for future in self._in_flight[key]:
                future.set_result(value)
            return value
        finally:
            del self._in_flight[key]

import numpy as np
from redis.commands.search.query import Query

# Store embeddings in Redis for semantic matching
def semantic_cache_get(query_embedding, threshold=0.92):
    # Redis vector search for semantically similar cached queries
    result = redis.ft('cache_index').search(
        Query('*=>[KNN 1 @embedding $vec AS score]')
        .sort_by('score')
        .return_fields('response', 'score')
        .dialect(2),
        {'vec': np.array(query_embedding, dtype=np.float32).tobytes()}
    )
    
    if result.total > 0 and (1 - result.docs[0].score) >= threshold:
        return result.docs[0].response
    return None

def semantic_cache_set(query, response, embedding, ttl=3600):
    redis.hset(f'semantic:{hash(query)}', mapping={
        'query': query,
        'response': response,
        'embedding': np.array(embedding, dtype=np.float32).tobytes()
    })
    redis.expire(f'semantic:{query}', ttl)

# Cache vector search results for common queries
def get_rag_context(query, top_k=5):
    # Check semantic cache first
    cached = semantic_cache_get(query_embedding(query))
    if cached:
        return cached
    
    # Perform vector search
    results = vector_store.similarity_search(query, k=top_k)
    
    # Cache for reuse
    semantic_cache_set(query, results, query_embedding(query))
    return results

def llm_inference_cached(prompt, model="gpt-4"):
    prompt_hash = hashlib.sha256(prompt.encode()).hexdigest()
    cache_key = f'llm:{model}:{prompt_hash}'
    
    cached = redis.get(cache_key)
    if cached:
        return json.loads(cached)
    
    # Expensive inference call
    response = call_llm_api(prompt, model)
    
    # Cache for 1 hour — adjust TTL based on prompt volatility
    redis.setex(cache_key, 3600, json.dumps(response))
    return response

# OpenAI prompt caching — reuse system prompt embeddings
response = client.chat.completions.create(
    model="gpt-4",
    messages=[
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_input}
    ],
    # System prompt is cached between requests
)

# Cache warming at deployment time
def warm_cache_for_deployment():
    hot_keys = [
        'config:feature_flags',
        'products:popular',
        'categories:top_level',
    ]
    for key in hot_keys:
        value = fetch_from_database(key)
        redis.setex(key, 3600, json.dumps(value))

# Periodic refresh-ahead for critical data
def refresh_ahead_loop():
    while True:
        for key in ['config:feature_flags', 'pricing:tiers']:
            ttl = redis.ttl(key)
            if ttl < 300:  # Refresh within 5 minutes of expiry
                value = fetch_from_database(key)
                redis.setex(key, 3600, json.dumps(value))
        time.sleep(60)

# Prometheus Redis exporter configuration
- name: redis
  rules:
    - alert: CacheHitRateLow
      expr: |
        (redis_keyspace_hits_total / 
        (redis_keyspace_hits_total + redis_keyspace_misses_total)) < 0.8
      for: 5m
      labels:
        severity: warning
      annotations:
        summary: "Cache hit rate below 80%"
        
    - alert: RedisMemoryHigh
      expr: redis_memory_used_bytes / redis_memory_max_bytes > 0.9
      for: 5m
      labels:
        severity: warning
      annotations:
        summary: "Redis memory usage above 90%"
        
    - alert: RedisEvictionsHigh
      expr: rate(redis_evicted_keys_total[5m]) > 10
      for: 5m
      labels:
        severity: warning
      annotations:
        summary: "High number of key evictions"

# Cache hit/miss tracking
class CacheMetrics:
    def __init__(self):
        self.hits = 0
        self.misses = 0
        self.errors = 0
    
    def record_hit(self):
        self.hits += 1
    
    def record_miss(self):
        self.misses += 1
    
    @property
    def hit_rate(self):
        total = self.hits + self.misses
        return self.hits / total if total > 0 else 0

metrics = CacheMetrics()

def get_cached(key):
    try:
        value = redis.get(key)
        if value:
            metrics.record_hit()
            return json.loads(value)
        metrics.record_miss()
        return None
    except Exception:
        metrics.errors += 1
        return None

# Memcached is simpler - no complex data types
import pymemcache

mc = pymemcache.Client(('localhost', 11211))

# Simple string caching only
mc.set('key', 'value', expire=3600)
value = mc.get('key')

import hashlib

# Encrypt sensitive values before caching
def cache_sensitive(key, value, ttl=3600):
    encrypted = encrypt(json.dumps(value), SECRET_KEY)
    redis.setex(key, ttl, encrypted)

# Use Redis 6+ ACLs for fine-grained access
# ACL: user caching on ~cache:* +@read +@write -@admin >secure_password

# Bad: Caching entire database rows
user = cache.get(f'user:{user_id}')  # Contains password hash!

# Good: Cache only what's needed
user_summary = cache.get(f'user:{user_id}:summary')

# Bad: No expiration
redis.set('config', json.dumps(config))

# Good: Reasonable TTL with refresh
redis.setex('config', 3600, json.dumps(config))

# Bad: Caching a simple primary-key lookup (already fast)
product = cache.get(f'product:{pid}')  # DB query was 2ms

# Good: Cache expensive joins and aggregations
sales_report = cache.get('sales:q2:report')  # DB query was 500ms

# Bad: Single key for all active users
redis.zadd('active_users', {user_id: time.time()})

# Good: Shard by hash of user_id (e.g., 16 shards)
shard = hash(user_id) % 16
redis.zadd(f'active_users:{shard}', {user_id: time.time()})

# Bad: Cache stampede
for i in range(100):  # 100 simultaneous requests
    user = get_user(123)  # All hit DB simultaneously

# Good: Request coalescing
lock = redis.lock('user:123:lock', timeout=5)
if lock.acquire():
    try:
        user = get_user(123)
    finally:
        lock.release()
else:
    # Wait for other request to populate cache
    time.sleep(0.1)
    user = get_user(123)

- alert: RedisMemoryWarning
  expr: redis_memory_used_bytes / redis_memory_max_bytes > 0.7
  for: 5m
  labels:
    severity: warning
  annotations:
    summary: "Redis memory usage above 70%"