Database Caching Strategies

# Installation
pip install redis

import redis
from datetime import timedelta

# Connect to Redis
cache = redis.Redis(
    host='localhost',
    port=6379,
    db=0,
    decode_responses=True  # Return strings instead of bytes
)

# Basic operations
cache.set('username', 'alice')
cache.set('user:1:email', 'alice@example.com')
cache.set('view_count', 0)

# Retrieve values
username = cache.get('username')  # 'alice'
email = cache.get('user:1:email')  # 'alice@example.com'

# Increment counter
cache.incr('view_count')  # Returns 1
cache.incr('view_count')  # Returns 2

# Result: Basic key-value operations with atomic increments

# Set with expiration time (auto-delete after TTL)
cache.setex('session:abc123', timedelta(hours=1), 'user_data')
# Result: Key expires in 1 hour, automatically deleted

# Alternative syntax (set + expire)
cache.set('temp:token', 'xyz789')
cache.expire('temp:token', 300)  # Expire in 300 seconds (5 minutes)

# Check remaining TTL
ttl = cache.ttl('session:abc123')  # Returns seconds remaining
print(f"Session expires in {ttl} seconds")

# Result: 3599 (just under 1 hour)

# Persist (remove expiration)
cache.persist('session:abc123')  # Key will never expire now

# Result: Keys automatically clean themselves up, no manual deletion needed

import redis
import psycopg2
import json

cache = redis.Redis(host='localhost', decode_responses=True)
db_conn = psycopg2.connect("dbname=myapp user=postgres")

def get_user(user_id):
    """Get user with cache-aside pattern."""
    cache_key = f"user:{user_id}"

    # Step 1: Try cache first
    cached_data = cache.get(cache_key)
    if cached_data:
        print("Cache HIT")
        return json.loads(cached_data)

    # Step 2: Cache miss - query database
    print("Cache MISS - querying database")
    cur = db_conn.cursor()
    cur.execute("SELECT id, username, email FROM users WHERE id = %s", (user_id,))
    row = cur.fetchone()

    if not row:
        return None

    user = {"id": row[0], "username": row[1], "email": row[2]}

    # Step 3: Populate cache with TTL
    cache.setex(cache_key, timedelta(hours=1), json.dumps(user))

    return user

# Usage
user1 = get_user(123)  # Cache MISS - querying database (10ms)
user2 = get_user(123)  # Cache HIT (0.5ms) - 20x faster!
user3 = get_user(123)  # Cache HIT (0.5ms)

# Result:
# First request: Slow (database query)
# Subsequent requests: Fast (cache hit)

# Problem: Cache expires at midnight, 1000 concurrent requests all miss cache
# All 1000 requests hit database simultaneously (stampede!)

import threading

def get_user_with_lock(user_id):
    """Prevent cache stampede with distributed lock."""
    cache_key = f"user:{user_id}"
    lock_key = f"lock:{cache_key}"

    # Try cache
    cached_data = cache.get(cache_key)
    if cached_data:
        return json.loads(cached_data)

    # Acquire lock (only one request rebuilds cache)
    lock_acquired = cache.set(lock_key, "1", nx=True, ex=5)  # 5-second lock

    if lock_acquired:
        try:
            # This request rebuilds cache
            user = query_database(user_id)
            cache.setex(cache_key, timedelta(hours=1), json.dumps(user))
            return user
        finally:
            cache.delete(lock_key)  # Release lock
    else:
        # Another request is rebuilding, wait and retry
        import time
        time.sleep(0.1)  # Wait 100ms
        return get_user_with_lock(user_id)  # Retry (should hit cache now)

def update_user(user_id, email):
    """Update user with write-around (invalidate cache)."""
    cache_key = f"user:{user_id}"

    # Step 1: Write to database
    cur = db_conn.cursor()
    cur.execute(
        "UPDATE users SET email = %s WHERE id = %s",
        (email, user_id)
    )
    db_conn.commit()

    # Step 2: Invalidate cache (delete stale entry)
    cache.delete(cache_key)

    # Result: Database is authoritative source
    # Next read will cache miss, fetch fresh data from DB

# Usage
update_user(123, 'newemail@example.com')
# Cache deleted, next get_user(123) will be slow but accurate

def update_user_write_through(user_id, email):
    """Update user with write-through (update cache)."""
    cache_key = f"user:{user_id}"

    # Step 1: Write to database
    cur = db_conn.cursor()
    cur.execute(
        "UPDATE users SET email = %s, updated_at = NOW() WHERE id = %s RETURNING *",
        (email, user_id)
    )
    row = cur.fetchone()
    db_conn.commit()

    # Step 2: Update cache with fresh data
    user = {"id": row[0], "username": row[1], "email": row[2]}
    cache.setex(cache_key, timedelta(hours=1), json.dumps(user))

    # Result: Cache has latest data, next read is fast

# Usage
update_user_write_through(123, 'newemail@example.com')
# Cache updated, next get_user(123) is instant (cache hit)

# Simplest: Let cache expire naturally

cache.setex('product:123', timedelta(minutes=15), json.dumps(product_data))

# Result:
# - Stale data possible for up to 15 minutes
# - No invalidation logic needed
# - Works well for data that changes infrequently

# TTL Guidelines:
# - User sessions: 30 minutes - 24 hours
# - Product catalog: 5-15 minutes
# - User profiles: 1-6 hours
# - Static content: 24 hours - 7 days
# - Real-time data: Don't cache, or use 10-30 seconds

# Invalidate cache when data changes

def update_product(product_id, new_price):
    """Update product and invalidate related cache entries."""

    # Update database and fetch category for cache invalidation
    cur = db_conn.cursor()
    cur.execute(
        "UPDATE products SET price = %s WHERE id = %s RETURNING category_id",
        (new_price, product_id)
    )
    row = cur.fetchone()
    db_conn.commit()

    category_id = row[0]

    # Invalidate all related cache keys
    cache.delete(f"product:{product_id}")                    # Individual product
    cache.delete(f"product:{product_id}:details")            # Product details page
    cache.delete(f"category:{category_id}:products")         # Category listing
    cache.delete("homepage:featured_products")               # Homepage cache

    # Result: All cached pages with this product are now stale
    # Next request will rebuild cache with fresh data

# Use Redis Sets to group related cache keys

def cache_with_tags(cache_key, data, tags, ttl=3600):
    """Cache data and associate with tags for bulk invalidation."""

    # Store data
    cache.setex(cache_key, ttl, json.dumps(data))

    # Add cache key to tag sets
    for tag in tags:
        cache.sadd(f"tag:{tag}", cache_key)
        cache.expire(f"tag:{tag}", ttl)  # Tag set expires with data

def invalidate_by_tag(tag):
    """Invalidate all cache entries with a specific tag."""
    tag_key = f"tag:{tag}"

    # Get all cache keys with this tag
    cache_keys = cache.smembers(tag_key)

    # Delete all tagged entries
    if cache_keys:
        cache.delete(*cache_keys)
        cache.delete(tag_key)  # Delete tag set itself

    print(f"Invalidated {len(cache_keys)} entries for tag '{tag}'")

# Usage
cache_with_tags(
    "product:123",
    product_data,
    tags=["product", "category:electronics", "brand:apple"],
    ttl=900
)

cache_with_tags(
    "product:456",
    product_data,
    tags=["product", "category:electronics", "brand:samsung"],
    ttl=900
)

# Invalidate all electronics category cache
invalidate_by_tag("category:electronics")
# Result: Invalidated 2 entries for tag 'category:electronics'
# Both products cleared from cache

import hashlib

def search_products(query, category=None, min_price=None, max_price=None):
    """Search products with query result caching."""

    # Generate cache key from search parameters (deterministic)
    cache_params = f"{query}:{category}:{min_price}:{max_price}"
    cache_key = f"search:{hashlib.md5(cache_params.encode()).hexdigest()}"

    # Try cache
    cached_results = cache.get(cache_key)
    if cached_results:
        print("Cache HIT - returning cached search results")
        return json.loads(cached_results)

    # Cache miss - run expensive query
    print("Cache MISS - executing database query")
    cur = db_conn.cursor()

    query_sql = """
        SELECT p.id, p.name, p.price, c.name AS category
        FROM products p
        JOIN categories c ON p.category_id = c.id
        WHERE p.name ILIKE %s
    """
    params = [f"%{query}%"]

    if category:
        query_sql += " AND c.name = %s"
        params.append(category)

    if min_price:
        query_sql += " AND p.price >= %s"
        params.append(min_price)

    if max_price:
        query_sql += " AND p.price <= %s"
        params.append(max_price)

    cur.execute(query_sql, params)
    results = [dict(zip(['id', 'name', 'price', 'category'], row))
               for row in cur.fetchall()]

    # Cache results for 5 minutes
    cache.setex(cache_key, 300, json.dumps(results))

    return results

# Usage
results1 = search_products("laptop", category="Electronics", min_price=500)
# Cache MISS - executing database query (50ms)

results2 = search_products("laptop", category="Electronics", min_price=500)
# Cache HIT - returning cached search results (1ms) - 50x faster!

# Use Redis Sorted Sets for leaderboards (no database!)

def update_user_score(user_id, score):
    """Update user score in leaderboard."""
    cache.zadd('leaderboard:global', {f"user:{user_id}": score})
    # Result: Score stored in Redis sorted set, O(log N) operation

def get_top_users(limit=10):
    """Get top N users from leaderboard."""
    # Get top users with scores (descending order)
    top_users = cache.zrevrange('leaderboard:global', 0, limit - 1, withscores=True)

    # Format results
    leaderboard = [
        {"user_id": user.split(':')[1], "score": int(score)}
        for user, score in top_users
    ]
    return leaderboard

# Usage
update_user_score(123, 9500)
update_user_score(456, 12000)
update_user_score(789, 8000)

top_10 = get_top_users(10)
# Result: [{'user_id': '456', 'score': 12000},
#          {'user_id': '123', 'score': 9500},
#          {'user_id': '789', 'score': 8000}]
# Retrieved in <1ms, no database query!

-- MySQL Query Cache (removed in MySQL 8.0)
-- Cached queries by exact SQL text match

SELECT * FROM users WHERE id = 1;  -- Cache MISS, query executed, result cached
SELECT * FROM users WHERE id = 1;  -- Cache HIT, instant result

-- Problem: Cache invalidated on ANY write to table
UPDATE users SET email = 'new@example.com' WHERE id = 999;
-- Result: Entire query cache for 'users' table CLEARED

-- Why it was removed:
-- - Too aggressive invalidation (any write clears all cached queries)
-- - Single mutex caused contention at high concurrency
-- - Application-level caching (Redis) is more flexible

-- PostgreSQL doesn't cache query results, but caches data pages in RAM

-- postgresql.conf
shared_buffers = 4GB          -- RAM for caching table/index pages
effective_cache_size = 12GB   -- OS + PG cache combined

-- How it works:
-- 1. First query reads from disk (slow)
-- 2. Pages loaded into shared_buffers (RAM)
-- 3. Subsequent queries hit RAM (fast)

-- Check cache hit rate
SELECT
    sum(heap_blks_read) AS disk_reads,
    sum(heap_blks_hit) AS cache_hits,
    sum(heap_blks_hit) / (sum(heap_blks_hit) + sum(heap_blks_read)) AS cache_hit_ratio
FROM pg_statio_user_tables;

-- Goal: >99% cache hit ratio
-- If <95%, increase shared_buffers