Amazon ElastiCache¶

import redis
import json
from datetime import timedelta

class RedisCache:
    def __init__(self, host, port=6379, password=None, ssl=True):
        self.client = redis.Redis(
            host=host,
            port=port,
            password=password,
            ssl=ssl,
            ssl_cert_reqs=None,
            decode_responses=True
        )

    def get(self, key):
        """Get value from cache"""
        try:
            value = self.client.get(key)
            return json.loads(value) if value else None
        except Exception as e:
            print(f"Cache get error: {e}")
            return None

    def set(self, key, value, ttl=3600):
        """Set value in cache with TTL"""
        try:
            serialized_value = json.dumps(value)
            self.client.setex(key, ttl, serialized_value)
            return True
        except Exception as e:
            print(f"Cache set error: {e}")
            return False

    def delete(self, key):
        """Delete key from cache"""
        try:
            return self.client.delete(key)
        except Exception as e:
            print(f"Cache delete error: {e}")
            return False

    def exists(self, key):
        """Check if key exists"""
        try:
            return self.client.exists(key)
        except Exception as e:
            print(f"Cache exists error: {e}")
            return False

# Usage example
cache = RedisCache(
    host='redis-prod-cluster.abc123.cache.amazonaws.com',
    password='SecureAuthToken123!'
)

# Caching database query results
def get_user_profile(user_id):
    cache_key = f"user_profile:{user_id}"

    # Try cache first
    profile = cache.get(cache_key)
    if profile:
        print("Cache hit")
        return profile

    # Cache miss - query database
    print("Cache miss - querying database")
    profile = database.get_user_profile(user_id)

    # Store in cache for 1 hour
    cache.set(cache_key, profile, ttl=3600)

    return profile

import redis
from rediscluster import RedisCluster

# Redis Cluster connection
startup_nodes = [
    {"host": "redis-cluster-001.abc123.cache.amazonaws.com", "port": "6379"},
    {"host": "redis-cluster-002.abc123.cache.amazonaws.com", "port": "6379"},
    {"host": "redis-cluster-003.abc123.cache.amazonaws.com", "port": "6379"}
]

cluster = RedisCluster(
    startup_nodes=startup_nodes,
    password='SecureAuthToken123!',
    ssl=True,
    ssl_cert_reqs=None,
    decode_responses=True,
    skip_full_coverage_check=True
)

# Distributed caching across cluster
def distributed_cache_example():
    # Data will be automatically distributed across cluster nodes
    for i in range(1000):
        cluster.set(f"key:{i}", f"value:{i}")

    # Retrieve data - Redis Cluster handles routing
    value = cluster.get("key:500")
    print(f"Retrieved: {value}")

# Using Redis data structures
def redis_data_structures():
    # Lists for queues
    cluster.lpush("task_queue", "task1", "task2", "task3")
    task = cluster.rpop("task_queue")

    # Sets for unique collections
    cluster.sadd("user_sessions", "session1", "session2")
    active_sessions = cluster.smembers("user_sessions")

    # Sorted sets for leaderboards
    cluster.zadd("leaderboard", {"player1": 100, "player2": 200})
    top_players = cluster.zrevrange("leaderboard", 0, 9, withscores=True)

    # Hash maps for objects
    cluster.hset("user:123", mapping={
        "name": "John Doe",
        "email": "john@example.com",
        "last_login": "2023-01-01"
    })
    user_data = cluster.hgetall("user:123")

from flask import Flask, session, request
import redis
import json
import uuid

app = Flask(__name__)

# Redis session store
redis_client = redis.Redis(
    host='redis-prod-cluster.abc123.cache.amazonaws.com',
    password='SecureAuthToken123!',
    ssl=True,
    decode_responses=True
)

class RedisSessionStore:
    def __init__(self, redis_client, ttl=3600):
        self.redis = redis_client
        self.ttl = ttl

    def create_session(self, user_id, session_data):
        session_id = str(uuid.uuid4())
        session_key = f"session:{session_id}"

        session_data.update({
            'user_id': user_id,
            'created_at': time.time()
        })

        self.redis.setex(
            session_key,
            self.ttl,
            json.dumps(session_data)
        )

        return session_id

    def get_session(self, session_id):
        session_key = f"session:{session_id}"
        session_data = self.redis.get(session_key)

        if session_data:
            return json.loads(session_data)
        return None

    def update_session(self, session_id, session_data):
        session_key = f"session:{session_id}"

        if self.redis.exists(session_key):
            self.redis.setex(
                session_key,
                self.ttl,
                json.dumps(session_data)
            )
            return True
        return False

    def delete_session(self, session_id):
        session_key = f"session:{session_id}"
        return self.redis.delete(session_key)

# Usage in Flask application
session_store = RedisSessionStore(redis_client)

@app.route('/login', methods=['POST'])
def login():
    user_id = authenticate_user(request.json)
    if user_id:
        session_data = {
            'user_id': user_id,
            'permissions': get_user_permissions(user_id)
        }
        session_id = session_store.create_session(user_id, session_data)
        return {'session_id': session_id}
    return {'error': 'Authentication failed'}, 401

@app.route('/profile')
def profile():
    session_id = request.headers.get('Session-ID')
    session_data = session_store.get_session(session_id)

    if session_data:
        user_id = session_data['user_id']
        return get_user_profile(user_id)

    return {'error': 'Invalid session'}, 401

# Application-level monitoring
import time
import logging

class CacheMonitor:
    def __init__(self, cache_client):
        self.cache = cache_client
        self.metrics = {
            'hits': 0,
            'misses': 0,
            'errors': 0,
            'total_time': 0,
            'operations': 0
        }

    def get_with_monitoring(self, key):
        start_time = time.time()
        try:
            value = self.cache.get(key)

            if value:
                self.metrics['hits'] += 1
            else:
                self.metrics['misses'] += 1

            self.metrics['operations'] += 1
            self.metrics['total_time'] += time.time() - start_time

            return value
        except Exception as e:
            self.metrics['errors'] += 1
            logging.error(f"Cache error: {e}")
            return None

    def get_stats(self):
        total_ops = self.metrics['operations']
        if total_ops == 0:
            return {}

        hit_rate = self.metrics['hits'] / total_ops
        avg_latency = self.metrics['total_time'] / total_ops * 1000  # ms

        return {
            'hit_rate': hit_rate,
            'miss_rate': 1 - hit_rate,
            'error_rate': self.metrics['errors'] / total_ops,
            'average_latency_ms': avg_latency,
            'total_operations': total_ops
        }

# Usage
monitor = CacheMonitor(redis_client)

# Use monitored cache operations
user_data = monitor.get_with_monitoring("user:123")

# Log statistics periodically
stats = monitor.get_stats()
logging.info(f"Cache stats: {stats}")