Redis State Management

A comprehensive skill for mastering Redis state management patterns in distributed systems. This skill covers caching strategies, session management, pub/sub messaging, distributed locks, data structures, and production-ready patterns using redis-py.

When to Use This Skill

Use this skill when:

• Implementing high-performance caching layers for web applications

• Managing user sessions in distributed environments

• Building real-time messaging and event distribution systems

• Coordinating distributed processes with locks and synchronization

• Storing and querying structured data with Redis data structures

• Optimizing application performance with Redis

• Scaling applications horizontally with shared state

• Implementing rate limiting, counters, and analytics

• Building microservices with Redis as a communication layer

• Managing temporary data with automatic expiration (TTL)

• Implementing leaderboards, queues, and real-time features

Core Concepts

Redis Fundamentals

Redis (Remote Dictionary Server) is an in-memory data structure store used as:

• Database: Persistent key-value storage

• Cache: High-speed data layer

• Message Broker: Pub/sub and stream messaging

• Session Store: Distributed session management

Key Characteristics:

• In-memory storage (microsecond latency)

• Optional persistence (RDB snapshots, AOF logs)

• Rich data structures beyond key-value

• Atomic operations on complex data types

• Built-in replication and clustering

• Pub/sub messaging support

• Lua scripting for complex operations

• Pipelining for batch operations

Redis Data Structures

Redis provides multiple data types for different use cases:

Strings: Simple key-value pairs, binary safe

• Use for: Cache values, counters, flags, JSON objects

• Max size: 512 MB

• Commands: SET, GET, INCR, APPEND

Hashes: Field-value maps (objects)

• Use for: User profiles, configuration objects, small entities

• Efficient for storing objects with multiple fields

• Commands: HSET, HGET, HMGET, HINCRBY

Lists: Ordered collections (linked lists)

• Use for: Queues, activity feeds, recent items

• Operations at head/tail are O(1)

• Commands: LPUSH, RPUSH, LPOP, RPOP, LRANGE

Sets: Unordered unique collections

• Use for: Tags, unique visitors, relationships

• Set operations: union, intersection, difference

• Commands: SADD, SMEMBERS, SISMEMBER, SINTER

Sorted Sets: Ordered sets with scores

• Use for: Leaderboards, time-series, priority queues

• Range queries by score or rank

• Commands: ZADD, ZRANGE, ZRANGEBYSCORE, ZRANK

Streams: Append-only logs with consumer groups

• Use for: Event sourcing, activity logs, message queues

• Built-in consumer group support

• Commands: XADD, XREAD, XREADGROUP

Connection Management

Connection Pools: Redis connections are expensive to create. Always use connection pools:

import redis

Connection pool (recommended)

pool = redis.ConnectionPool(host='localhost', port=6379, db=0, max_connections=10) r = redis.Redis(connection_pool=pool)

Direct connection (avoid in production)

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

Best Practices:

• Use connection pools for all applications

• Set appropriate max_connections based on workload

• Enable decode_responses=True for string data

• Configure socket_timeout and socket_keepalive

• Handle connection errors with retries

Data Persistence

Redis offers two persistence mechanisms:

RDB (Redis Database): Point-in-time snapshots

• Compact binary format

• Fast restart times

• Lower disk I/O

• Potential data loss between snapshots

AOF (Append-Only File): Log of write operations

• Better durability (fsync policies)

• Larger files, slower restarts

• Can be automatically rewritten/compacted

• Minimal data loss potential

Hybrid Approach: RDB + AOF for best of both worlds

RESP 3 Protocol

Redis Serialization Protocol version 3 offers:

• Client-side caching support

• Better data type support

• Push notifications

• Performance improvements

import redis from redis.cache import CacheConfig

Enable RESP3 with client-side caching

r = redis.Redis(host='localhost', port=6379, protocol=3, cache_config=CacheConfig())

Caching Strategies

Cache-Aside (Lazy Loading)

Pattern: Application checks cache first, loads from database on miss

import redis import json from typing import Optional, Dict, Any

r = redis.Redis(decode_responses=True)

def get_user(user_id: int) -> Optional[Dict[str, Any]]: """Cache-aside pattern for user data.""" cache_key = f"user:{user_id}"

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

# Cache miss - load from database
user_data = database.get_user(user_id)  # Your DB query
if user_data:
    # Store in cache with 1 hour TTL
    r.setex(cache_key, 3600, json.dumps(user_data))

return user_data

Advantages:

• Only requested data is cached (efficient memory usage)

• Cache failures don't break the application

• Simple to implement

Disadvantages:

• Cache miss penalty (latency spike)

• Thundering herd on popular items

• Stale data until cache expiration

Write-Through Cache

Pattern: Write to cache and database simultaneously

def update_user(user_id: int, user_data: Dict[str, Any]) -> bool: """Write-through pattern for user updates.""" cache_key = f"user:{user_id}"

# Write to database first
success = database.update_user(user_id, user_data)

if success:
    # Update cache immediately
    r.setex(cache_key, 3600, json.dumps(user_data))

return success

Advantages:

• Cache always consistent with database

• No read penalty for recently written data

Disadvantages:

• Write latency increases

• Unused data may be cached

• Extra cache write overhead

Write-Behind (Write-Back) Cache

Pattern: Write to cache immediately, sync to database asynchronously

import redis import json from queue import Queue from threading import Thread

r = redis.Redis(decode_responses=True) write_queue = Queue()

def async_writer(): """Background worker to sync cache to database.""" while True: user_id, user_data = write_queue.get() try: database.update_user(user_id, user_data) except Exception as e: # Log error, potentially retry print(f"Failed to write user {user_id}: {e}") finally: write_queue.task_done()

Start background writer

Thread(target=async_writer, daemon=True).start()

def update_user_fast(user_id: int, user_data: Dict[str, Any]): """Write-behind pattern for fast writes.""" cache_key = f"user:{user_id}"

# Write to cache immediately (fast)
r.setex(cache_key, 3600, json.dumps(user_data))

# Queue database write (async)
write_queue.put((user_id, user_data))

Advantages:

• Minimal write latency

• Can batch database writes

• Handles write spikes

Disadvantages:

• Risk of data loss if cache fails

• Complex error handling

• Consistency challenges

Cache Invalidation Strategies

Time-based Expiration (TTL):

Set key with expiration

r.setex("session:abc123", 1800, session_data) # 30 minutes

Or set TTL on existing key

r.expire("user:profile:123", 3600) # 1 hour

Check remaining TTL

ttl = r.ttl("user:profile:123")

Event-based Invalidation:

def update_product(product_id: int, product_data: dict): """Invalidate cache on update.""" # Update database database.update_product(product_id, product_data)

# Invalidate related caches
r.delete(f"product:{product_id}")
r.delete(f"product_list:category:{product_data['category']}")
r.delete("products:featured")

Pattern-based Invalidation:

Delete all keys matching pattern

def invalidate_user_cache(user_id: int): """Invalidate all cache entries for a user.""" pattern = f"user:{user_id}:*"

# Find and delete matching keys
for key in r.scan_iter(match=pattern, count=100):
    r.delete(key)

Cache Stampede Prevention

Problem: Multiple requests simultaneously miss cache and query database

Solution 1: Probabilistic Early Expiration

import time import random

def get_with_early_expiration(key: str, ttl: int = 3600, beta: float = 1.0): """Prevent stampede with probabilistic early recomputation.""" value = r.get(key)

if value is None:
    # Cache miss - compute and cache
    value = compute_value(key)
    r.setex(key, ttl, value)
    return value

# Check if we should recompute early
current_time = time.time()
delta = current_time - float(r.get(f"{key}:timestamp") or 0)
expiry = ttl * random.random() * beta

if delta > expiry:
    # Recompute in background
    value = compute_value(key)
    r.setex(key, ttl, value)
    r.set(f"{key}:timestamp", current_time)

return value

Solution 2: Locking

from contextlib import contextmanager

@contextmanager def cache_lock(key: str, timeout: int = 10): """Acquire lock for cache computation.""" lock_key = f"{key}:lock" identifier = str(time.time())

# Try to acquire lock
if r.set(lock_key, identifier, nx=True, ex=timeout):
    try:
        yield True
    finally:
        # Release lock
        if r.get(lock_key) == identifier:
            r.delete(lock_key)
else:
    yield False

def get_with_lock(key: str): """Use lock to prevent stampede.""" value = r.get(key)

if value is None:
    with cache_lock(key) as acquired:
        if acquired:
            # We got the lock - compute value
            value = compute_value(key)
            r.setex(key, 3600, value)
        else:
            # Someone else is computing - wait and retry
            time.sleep(0.1)
            value = r.get(key) or compute_value(key)

return value

Session Management

Distributed Session Storage

Basic Session Management:

import redis import json import uuid from datetime import datetime, timedelta

r = redis.Redis(decode_responses=True)

class SessionManager: def init(self, ttl: int = 1800): """Session manager with Redis backend.

    Args:
        ttl: Session timeout in seconds (default 30 minutes)
    """
    self.ttl = ttl

def create_session(self, user_id: int, data: dict = None) -> str:
    """Create new session and return session ID."""
    session_id = str(uuid.uuid4())
    session_key = f"session:{session_id}"

    session_data = {
        "user_id": user_id,
        "created_at": datetime.utcnow().isoformat(),
        "data": data or {}
    }

    r.setex(session_key, self.ttl, json.dumps(session_data))
    return session_id

def get_session(self, session_id: str) -> dict:
    """Retrieve session data and refresh TTL."""
    session_key = f"session:{session_id}"
    session_data = r.get(session_key)

    if session_data:
        # Refresh TTL on access (sliding expiration)
        r.expire(session_key, self.ttl)
        return json.loads(session_data)

    return None

def update_session(self, session_id: str, data: dict) -> bool:
    """Update session data."""
    session_key = f"session:{session_id}"
    session_data = self.get_session(session_id)

    if session_data:
        session_data["data"].update(data)
        r.setex(session_key, self.ttl, json.dumps(session_data))
        return True

    return False

def delete_session(self, session_id: str) -> bool:
    """Delete session (logout)."""
    session_key = f"session:{session_id}"
    return r.delete(session_key) > 0

Session with Hash Storage

More efficient for session objects:

class HashSessionManager: """Session manager using Redis hashes for better performance."""

def __init__(self, ttl: int = 1800):
    self.ttl = ttl

def create_session(self, user_id: int, **kwargs) -> str:
    """Create session using hash."""
    session_id = str(uuid.uuid4())
    session_key = f"session:{session_id}"

    # Store as hash for efficient field access
    session_fields = {
        "user_id": str(user_id),
        "created_at": datetime.utcnow().isoformat(),
        **{k: str(v) for k, v in kwargs.items()}
    }

    r.hset(session_key, mapping=session_fields)
    r.expire(session_key, self.ttl)

    return session_id

def get_field(self, session_id: str, field: str) -> str:
    """Get single session field efficiently."""
    session_key = f"session:{session_id}"
    value = r.hget(session_key, field)

    if value:
        r.expire(session_key, self.ttl)  # Refresh TTL

    return value

def set_field(self, session_id: str, field: str, value: str) -> bool:
    """Update single session field."""
    session_key = f"session:{session_id}"

    if r.exists(session_key):
        r.hset(session_key, field, value)
        r.expire(session_key, self.ttl)
        return True

    return False

def get_all(self, session_id: str) -> dict:
    """Get all session fields."""
    session_key = f"session:{session_id}"
    data = r.hgetall(session_key)

    if data:
        r.expire(session_key, self.ttl)

    return data

User Activity Tracking

def track_user_activity(user_id: int, action: str): """Track user activity with automatic expiration.""" activity_key = f"user:{user_id}:activity" timestamp = datetime.utcnow().isoformat()

# Add activity to list
r.lpush(activity_key, json.dumps({"action": action, "timestamp": timestamp}))

# Keep only last 100 activities
r.ltrim(activity_key, 0, 99)

# Set expiration (30 days)
r.expire(activity_key, 2592000)

def get_recent_activity(user_id: int, limit: int = 10) -> list: """Get recent user activities.""" activity_key = f"user:{user_id}:activity" activities = r.lrange(activity_key, 0, limit - 1)

return [json.loads(a) for a in activities]

Pub/Sub Patterns

Basic Publisher/Subscriber

Publisher:

import redis

r = redis.Redis(decode_responses=True)

def publish_event(channel: str, message: dict): """Publish event to channel.""" import json r.publish(channel, json.dumps(message))

Example usage

publish_event("notifications", { "type": "user_signup", "user_id": 12345, "timestamp": datetime.utcnow().isoformat() })

Subscriber:

import redis import json

def handle_message(message): """Process received message.""" data = json.loads(message['data']) print(f"Received: {data}")

Initialize pubsub

r = redis.Redis(decode_responses=True) p = r.pubsub()

Subscribe to channels

p.subscribe('notifications', 'alerts')

Listen for messages

for message in p.listen(): if message['type'] == 'message': handle_message(message)

Pattern-Based Subscriptions

Subscribe to multiple channels with patterns

p = r.pubsub() p.psubscribe('user:', 'notification:')

Get messages from pattern subscriptions

for message in p.listen(): if message['type'] == 'pmessage': channel = message['channel'] pattern = message['pattern'] data = message['data'] print(f"Pattern {pattern} matched {channel}: {data}")

Async Pub/Sub with Background Thread

import redis import time

r = redis.Redis(decode_responses=True) p = r.pubsub()

def message_handler(message): """Handle messages in background thread.""" print(f"Handler received: {message['data']}")

Subscribe with handler

p.subscribe(**{'notifications': message_handler, 'alerts': message_handler})

Run in background thread

thread = p.run_in_thread(sleep_time=0.001)

Publish some messages

r.publish('notifications', 'Hello!') r.publish('alerts', 'Warning!')

time.sleep(1)

Stop background thread

thread.stop()

Async Pub/Sub with asyncio

import asyncio import redis.asyncio as redis

async def reader(channel: redis.client.PubSub): """Async message reader.""" while True: message = await channel.get_message(ignore_subscribe_messages=True, timeout=None) if message is not None: print(f"Received: {message}")

        # Stop on specific message
        if message["data"].decode() == "STOP":
            break

async def pubsub_example(): """Async pub/sub example.""" r = await redis.from_url("redis://localhost")

async with r.pubsub() as pubsub:
    # Subscribe to channels
    await pubsub.subscribe("channel:1", "channel:2")

    # Create reader task
    reader_task = asyncio.create_task(reader(pubsub))

    # Publish messages
    await r.publish("channel:1", "Hello")
    await r.publish("channel:2", "World")
    await r.publish("channel:1", "STOP")

    # Wait for reader to finish
    await reader_task

await r.close()

Run async example

asyncio.run(pubsub_example())

Sharded Pub/Sub (Redis 7.0+)

from redis.cluster import RedisCluster, ClusterNode

Connect to cluster

rc = RedisCluster(startup_nodes=[ ClusterNode('localhost', 6379), ClusterNode('localhost', 6380) ])

Create sharded pubsub

p = rc.pubsub() p.ssubscribe('foo')

Get message from specific node

message = p.get_sharded_message(target_node=ClusterNode('localhost', 6379))

Distributed Locks

Simple Lock Implementation

import redis import time import uuid

class RedisLock: """Simple distributed lock using Redis."""

def __init__(self, redis_client: redis.Redis, key: str, timeout: int = 10):
    self.redis = redis_client
    self.key = f"lock:{key}"
    self.timeout = timeout
    self.identifier = str(uuid.uuid4())

def acquire(self, blocking: bool = True, timeout: float = None) -> bool:
    """Acquire lock."""
    end_time = time.time() + (timeout or self.timeout)

    while True:
        # Try to set lock with NX (only if not exists) and EX (expiration)
        if self.redis.set(self.key, self.identifier, nx=True, ex=self.timeout):
            return True

        if not blocking:
            return False

        if timeout and time.time() > end_time:
            return False

        # Wait before retry
        time.sleep(0.01)

def release(self) -> bool:
    """Release lock only if we own it."""
    # Use Lua script for atomic check-and-delete
    lua_script = """
    if redis.call("get", KEYS[1]) == ARGV[1] then
        return redis.call("del", KEYS[1])
    else
        return 0
    end
    """

    result = self.redis.eval(lua_script, 1, self.key, self.identifier)
    return result == 1

def __enter__(self):
    """Context manager support."""
    self.acquire()
    return self

def __exit__(self, exc_type, exc_val, exc_tb):
    """Context manager cleanup."""
    self.release()

Usage example

r = redis.Redis() lock = RedisLock(r, "resource:123", timeout=5)

with lock: # Critical section - only one process at a time print("Processing resource 123") process_resource()

Advanced Lock with Auto-Renewal

import threading

class RenewableLock: """Distributed lock with automatic renewal."""

def __init__(self, redis_client: redis.Redis, key: str, timeout: int = 10):
    self.redis = redis_client
    self.key = f"lock:{key}"
    self.timeout = timeout
    self.identifier = str(uuid.uuid4())
    self.renewal_thread = None
    self.stop_renewal = threading.Event()

def _renew_lock(self):
    """Background task to renew lock."""
    while not self.stop_renewal.is_set():
        time.sleep(self.timeout / 3)  # Renew at 1/3 of timeout

        # Renew only if we still own the lock
        lua_script = """
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("expire", KEYS[1], ARGV[2])
        else
            return 0
        end
        """

        result = self.redis.eval(lua_script, 1, self.key,
                               self.identifier, self.timeout)

        if result == 0:
            # We lost the lock
            self.stop_renewal.set()

def acquire(self, blocking: bool = True, timeout: float = None) -> bool:
    """Acquire lock and start auto-renewal."""
    if self.redis.set(self.key, self.identifier, nx=True, ex=self.timeout):
        # Start renewal thread
        self.stop_renewal.clear()
        self.renewal_thread = threading.Thread(target=self._renew_lock, daemon=True)
        self.renewal_thread.start()
        return True

    return False

def release(self) -> bool:
    """Release lock and stop renewal."""
    self.stop_renewal.set()

    lua_script = """
    if redis.call("get", KEYS[1]) == ARGV[1] then
        return redis.call("del", KEYS[1])
    else
        return 0
    end
    """

    result = self.redis.eval(lua_script, 1, self.key, self.identifier)
    return result == 1

Redlock Algorithm (Multiple Redis Instances)

class Redlock: """Redlock algorithm for distributed locking across multiple Redis instances."""

def __init__(self, redis_instances: list):
    """
    Args:
        redis_instances: List of Redis client connections
    """
    self.instances = redis_instances
    self.quorum = len(redis_instances) // 2 + 1

def acquire(self, resource: str, ttl: int = 10000) -> tuple:
    """
    Acquire lock across multiple Redis instances.

    Returns:
        (success: bool, lock_identifier: str)
    """
    identifier = str(uuid.uuid4())
    start_time = int(time.time() * 1000)

    # Try to acquire lock on all instances
    acquired = 0
    for instance in self.instances:
        try:
            if instance.set(f"lock:{resource}", identifier,
                          nx=True, px=ttl):
                acquired += 1
        except Exception:
            pass

    # Calculate elapsed time
    elapsed = int(time.time() * 1000) - start_time
    validity_time = ttl - elapsed - 100  # drift compensation

    # Check if we got quorum
    if acquired >= self.quorum and validity_time > 0:
        return True, identifier
    else:
        # Release locks if we didn't get quorum
        self._release_all(resource, identifier)
        return False, None

def _release_all(self, resource: str, identifier: str):
    """Release lock on all instances."""
    lua_script = """
    if redis.call("get", KEYS[1]) == ARGV[1] then
        return redis.call("del", KEYS[1])
    else
        return 0
    end
    """

    for instance in self.instances:
        try:
            instance.eval(lua_script, 1, f"lock:{resource}", identifier)
        except Exception:
            pass

Data Structures and Operations

Working with Hashes

User profile storage

def save_user_profile(user_id: int, profile: dict): """Save user profile as hash.""" key = f"user:profile:{user_id}" r.hset(key, mapping=profile) r.expire(key, 86400) # 24 hour TTL

def get_user_profile(user_id: int) -> dict: """Get complete user profile.""" key = f"user:profile:{user_id}" return r.hgetall(key)

def update_user_field(user_id: int, field: str, value: str): """Update single profile field.""" key = f"user:profile:{user_id}" r.hset(key, field, value)

Example usage

save_user_profile(123, { "username": "alice", "email": "alice@example.com", "age": "30" })

Atomic increment

r.hincrby("user:profile:123", "login_count", 1)

Working with Lists

Job queue implementation

def enqueue_job(queue_name: str, job_data: dict): """Add job to queue.""" key = f"queue:{queue_name}" r.rpush(key, json.dumps(job_data))

def dequeue_job(queue_name: str, timeout: int = 0) -> dict: """Get job from queue (blocking).""" key = f"queue:{queue_name}"

if timeout > 0:
    # Blocking pop with timeout
    result = r.blpop(key, timeout=timeout)
    if result:
        _, job_data = result
        return json.loads(job_data)
else:
    # Non-blocking pop
    job_data = r.lpop(key)
    if job_data:
        return json.loads(job_data)

return None

Activity feed

def add_to_feed(user_id: int, activity: dict): """Add activity to user feed.""" key = f"feed:{user_id}" r.lpush(key, json.dumps(activity)) r.ltrim(key, 0, 99) # Keep only latest 100 items r.expire(key, 604800) # 7 days

def get_feed(user_id: int, start: int = 0, end: int = 19) -> list: """Get user feed with pagination.""" key = f"feed:{user_id}" items = r.lrange(key, start, end) return [json.loads(item) for item in items]

Working with Sets

Tags and relationships

def add_tags(item_id: int, tags: list): """Add tags to item.""" key = f"item:{item_id}:tags" r.sadd(key, *tags)

def get_tags(item_id: int) -> set: """Get all tags for item.""" key = f"item:{item_id}:tags" return r.smembers(key)

def find_items_with_all_tags(tags: list) -> set: """Find items having all specified tags.""" keys = [f"item:*:tags" for _ in tags] # This is simplified - in practice, you'd need to track item IDs differently return r.sinter(*keys)

Online users tracking

def user_online(user_id: int): """Mark user as online.""" r.sadd("users:online", user_id) r.expire(f"user:{user_id}:heartbeat", 60)

def user_offline(user_id: int): """Mark user as offline.""" r.srem("users:online", user_id)

def get_online_users() -> set: """Get all online users.""" return r.smembers("users:online")

def get_online_count() -> int: """Get count of online users.""" return r.scard("users:online")

Working with Sorted Sets

Leaderboard implementation

def update_score(leaderboard: str, user_id: int, score: float): """Update user score in leaderboard.""" key = f"leaderboard:{leaderboard}" r.zadd(key, {user_id: score})

def get_leaderboard(leaderboard: str, start: int = 0, end: int = 9) -> list: """Get top players (descending order).""" key = f"leaderboard:{leaderboard}" # ZREVRANGE for descending order (highest scores first) return r.zrevrange(key, start, end, withscores=True)

def get_user_rank(leaderboard: str, user_id: int) -> int: """Get user's rank (0-indexed).""" key = f"leaderboard:{leaderboard}" # ZREVRANK for descending rank rank = r.zrevrank(key, user_id) return rank if rank is not None else -1

def get_user_score(leaderboard: str, user_id: int) -> float: """Get user's score.""" key = f"leaderboard:{leaderboard}" score = r.zscore(key, user_id) return score if score is not None else 0.0

def get_score_range(leaderboard: str, min_score: float, max_score: float) -> list: """Get users within score range.""" key = f"leaderboard:{leaderboard}" return r.zrangebyscore(key, min_score, max_score, withscores=True)

Time-based sorted set (activity stream)

def add_activity(user_id: int, activity: str): """Add timestamped activity.""" key = f"user:{user_id}:activities" timestamp = time.time() r.zadd(key, {activity: timestamp})

# Keep only last 24 hours
cutoff = timestamp - 86400
r.zremrangebyscore(key, '-inf', cutoff)

def get_recent_activities(user_id: int, count: int = 10) -> list: """Get recent activities.""" key = f"user:{user_id}:activities" # Get most recent (highest timestamps) return r.zrevrange(key, 0, count - 1, withscores=True)

Working with Streams

Event stream

def add_event(stream_key: str, event_data: dict) -> str: """Add event to stream.""" # Returns auto-generated ID (timestamp-sequence) event_id = r.xadd(stream_key, event_data) return event_id

def read_events(stream_key: str, count: int = 10, start_id: str = '0') -> list: """Read events from stream.""" events = r.xread({stream_key: start_id}, count=count)

# events format: [(stream_name, [(id, data), (id, data), ...])]
if events:
    _, event_list = events[0]
    return event_list

return []

Consumer groups

def create_consumer_group(stream_key: str, group_name: str): """Create consumer group for stream.""" try: r.xgroup_create(name=stream_key, groupname=group_name, id='0') except redis.ResponseError as e: if "BUSYGROUP" not in str(e): raise

def read_from_group(stream_key: str, group_name: str, consumer_name: str, count: int = 10) -> list: """Read events as consumer in group.""" # Read new messages with '>' events = r.xreadgroup( groupname=group_name, consumername=consumer_name, streams={stream_key: '>'}, count=count, block=5000 # 5 second timeout )

if events:
    _, event_list = events[0]
    return event_list

return []

def acknowledge_event(stream_key: str, group_name: str, event_id: str): """Acknowledge processed event.""" r.xack(stream_key, group_name, event_id)

Example: Processing events with consumer group

def process_events(stream_key: str, group_name: str, consumer_name: str): """Process events from stream.""" create_consumer_group(stream_key, group_name)

while True:
    events = read_from_group(stream_key, group_name, consumer_name, count=10)

    for event_id, event_data in events:
        try:
            # Process event
            process_event(event_data)

            # Acknowledge successful processing
            acknowledge_event(stream_key, group_name, event_id)
        except Exception as e:
            print(f"Failed to process event {event_id}: {e}")
            # Event remains unacknowledged for retry

Performance Optimization

Pipelining for Batch Operations

Without pipelining (slow - multiple round trips)

for i in range(1000): r.set(f"key:{i}", f"value:{i}")

With pipelining (fast - single round trip)

pipe = r.pipeline() for i in range(1000): pipe.set(f"key:{i}", f"value:{i}") results = pipe.execute()

Pipelining with reads

pipe = r.pipeline() for i in range(100): pipe.get(f"key:{i}") values = pipe.execute()

Builder pattern with pipeline

class DataLoader: def init(self): self.pipeline = r.pipeline()

def add_user(self, user_id: int, user_data: dict):
    """Add user data."""
    self.pipeline.hset(f"user:{user_id}", mapping=user_data)
    return self

def add_to_set(self, set_name: str, value: str):
    """Add to set."""
    self.pipeline.sadd(set_name, value)
    return self

def execute(self):
    """Execute all pipelined commands."""
    return self.pipeline.execute()

Usage

loader = DataLoader() results = (loader .add_user(1, {"name": "Alice", "email": "alice@example.com"}) .add_user(2, {"name": "Bob", "email": "bob@example.com"}) .add_to_set("active_users", "1") .add_to_set("active_users", "2") .execute())

Transactions with WATCH

Optimistic locking with WATCH

def transfer_credits(from_user: int, to_user: int, amount: int) -> bool: """Transfer credits between users with optimistic locking."""

with r.pipeline() as pipe:
    while True:
        try:
            # Watch the keys we're going to modify
            pipe.watch(f"user:{from_user}:credits", f"user:{to_user}:credits")

            # Get current values
            from_credits = int(pipe.get(f"user:{from_user}:credits") or 0)
            to_credits = int(pipe.get(f"user:{to_user}:credits") or 0)

            # Check if transfer is possible
            if from_credits < amount:
                pipe.unwatch()
                return False

            # Start transaction
            pipe.multi()
            pipe.set(f"user:{from_user}:credits", from_credits - amount)
            pipe.set(f"user:{to_user}:credits", to_credits + amount)

            # Execute transaction
            pipe.execute()
            return True

        except redis.WatchError:
            # Key was modified by another client - retry
            continue

Lua scripts for atomic operations

increment_script = """ local current = redis.call('GET', KEYS[1]) if not current then current = 0 end local new_val = tonumber(current) + tonumber(ARGV[1]) redis.call('SET', KEYS[1], new_val) return new_val """

Register and use Lua script

increment = r.register_script(increment_script) new_value = increment(keys=['counter:views'], args=[1])

Lua Scripts for Complex Operations

Rate limiting with Lua

rate_limit_script = """ local key = KEYS[1] local limit = tonumber(ARGV[1]) local window = tonumber(ARGV[2]) local current = redis.call('INCR', key)

if current == 1 then redis.call('EXPIRE', key, window) end

if current > limit then return 0 else return 1 end """

rate_limiter = r.register_script(rate_limit_script)

def is_allowed(user_id: int, limit: int = 100, window: int = 60) -> bool: """Check if user is within rate limit.""" key = f"rate_limit:{user_id}" result = rate_limiter(keys=[key], args=[limit, window]) return result == 1

Get-or-set pattern with Lua

get_or_set_script = """ local value = redis.call('GET', KEYS[1]) if value then return value else redis.call('SET', KEYS[1], ARGV[1]) redis.call('EXPIRE', KEYS[1], ARGV[2]) return ARGV[1] end """

get_or_set = r.register_script(get_or_set_script)

def get_or_compute(key: str, compute_fn, ttl: int = 3600): """Get value from cache or compute and cache it.""" value = get_or_set(keys=[key], args=["COMPUTING", ttl])

if value == "__COMPUTING__":
    # We set the placeholder - compute the real value
    computed = compute_fn()
    r.setex(key, ttl, computed)
    return computed

return value

Production Patterns

High Availability with Sentinel

from redis.sentinel import Sentinel

Connect to Sentinel

sentinel = Sentinel([ ('sentinel1', 26379), ('sentinel2', 26379), ('sentinel3', 26379) ], socket_timeout=0.5)

Get master connection

master = sentinel.master_for('mymaster', socket_timeout=0.5)

Get replica connection (for read-only operations)

replica = sentinel.slave_for('mymaster', socket_timeout=0.5)

Use master for writes

master.set('key', 'value')

Use replica for reads (optional, for load distribution)

value = replica.get('key')

Async Redis with asyncio

import asyncio import redis.asyncio as redis

async def async_redis_operations(): """Async Redis operations example.""" # Create async connection r = await redis.from_url("redis://localhost")

try:
    # Async operations
    await r.set("async_key", "async_value")
    value = await r.get("async_key")
    print(f"Value: {value}")

    # Async pipeline
    async with r.pipeline(transaction=True) as pipe:
        await pipe.set("key1", "value1")
        await pipe.set("key2", "value2")
        await pipe.get("key1")
        results = await pipe.execute()

    print(f"Pipeline results: {results}")

finally:
    await r.close()

Run async operations

asyncio.run(async_redis_operations())

Connection Pool Configuration

Production-ready connection pool

pool = redis.ConnectionPool( host='localhost', port=6379, db=0, max_connections=50, # Max pool size socket_timeout=5, # Socket timeout socket_connect_timeout=5, # Connection timeout socket_keepalive=True, # Keep TCP connection alive socket_keepalive_options={ socket.TCP_KEEPIDLE: 60, socket.TCP_KEEPINTVL: 10, socket.TCP_KEEPCNT: 3 }, retry_on_timeout=True, # Retry on timeout health_check_interval=30, # Health check every 30s decode_responses=True # Auto-decode bytes to strings )

r = redis.Redis(connection_pool=pool)

Error Handling and Resilience

import redis from redis.exceptions import ConnectionError, TimeoutError import time

class ResilientRedisClient: """Redis client with retry logic and circuit breaker."""

def __init__(self, max_retries: int = 3, backoff: float = 0.1):
    self.redis = redis.Redis(
        host='localhost',
        port=6379,
        socket_timeout=5,
        retry_on_timeout=True
    )
    self.max_retries = max_retries
    self.backoff = backoff

def get_with_retry(self, key: str, default=None):
    """Get value with exponential backoff retry."""
    for attempt in range(self.max_retries):
        try:
            return self.redis.get(key) or default
        except (ConnectionError, TimeoutError) as e:
            if attempt == self.max_retries - 1:
                # Log error and return default
                print(f"Redis error after {self.max_retries} attempts: {e}")
                return default

            # Exponential backoff
            wait_time = self.backoff * (2 ** attempt)
            time.sleep(wait_time)

def set_with_retry(self, key: str, value: str, ttl: int = None) -> bool:
    """Set value with retry logic."""
    for attempt in range(self.max_retries):
        try:
            if ttl:
                return self.redis.setex(key, ttl, value)
            else:
                return self.redis.set(key, value)
        except (ConnectionError, TimeoutError) as e:
            if attempt == self.max_retries - 1:
                print(f"Redis error after {self.max_retries} attempts: {e}")
                return False

            wait_time = self.backoff * (2 ** attempt)
            time.sleep(wait_time)

Monitoring and Metrics

def get_redis_info(section: str = None) -> dict: """Get Redis server information.""" return r.info(section=section)

def monitor_memory_usage(): """Monitor Redis memory usage.""" info = r.info('memory')

used_memory = info['used_memory_human']
peak_memory = info['used_memory_peak_human']
memory_fragmentation = info['mem_fragmentation_ratio']

print(f"Used Memory: {used_memory}")
print(f"Peak Memory: {peak_memory}")
print(f"Fragmentation Ratio: {memory_fragmentation}")

return info

def monitor_stats(): """Monitor Redis statistics.""" info = r.info('stats')

total_connections = info['total_connections_received']
total_commands = info['total_commands_processed']
ops_per_sec = info['instantaneous_ops_per_sec']

print(f"Total Connections: {total_connections}")
print(f"Total Commands: {total_commands}")
print(f"Ops/sec: {ops_per_sec}")

return info

def get_slow_log(count: int = 10): """Get slow query log.""" slow_log = r.slowlog_get(count)

for entry in slow_log:
    print(f"Command: {entry['command']}")
    print(f"Duration: {entry['duration']} microseconds")
    print(f"Time: {entry['start_time']}")
    print("---")

return slow_log

Best Practices

Key Naming Conventions

Use consistent, hierarchical naming:

Good naming patterns

user:123:profile # User profile data user:123:sessions:abc # User session cache:product:456 # Cached product queue:emails:pending # Email queue lock:resource:789 # Resource lock counter:api:requests:daily # Daily API request counter leaderboard:global:score # Global leaderboard

Avoid

u123 # Too cryptic user_profile_123 # Underscores less common 123:user # Wrong hierarchy

Memory Management

Set TTL on all temporary data

r.setex("temp:data", 3600, value) # Expires in 1 hour

Limit collection sizes

r.lpush("activity_log", entry) r.ltrim("activity_log", 0, 999) # Keep only 1000 items

Use appropriate data structures

Hash is more memory-efficient than multiple keys

r.hset("user:123", mapping={"name": "Alice", "email": "alice@example.com"})

vs

r.set("user:123:name", "Alice") r.set("user:123:email", "alice@example.com")

Monitor memory usage

if r.info('memory')['used_memory'] > threshold: # Implement eviction or cleanup cleanup_old_data()

Security

Use authentication

r = redis.Redis( host='localhost', port=6379, password='your-secure-password', username='your-username' # Redis 6+ )

Use SSL/TLS for production

pool = redis.ConnectionPool( host='redis.example.com', port=6380, connection_class=redis.SSLConnection, ssl_cert_reqs='required', ssl_ca_certs='/path/to/ca-cert.pem' )

Credential provider pattern

from redis import UsernamePasswordCredentialProvider

creds_provider = UsernamePasswordCredentialProvider("username", "password") r = redis.Redis( host="localhost", port=6379, credential_provider=creds_provider )

Testing

import fakeredis import pytest

@pytest.fixture def redis_client(): """Provide fake Redis client for testing.""" return fakeredis.FakeRedis(decode_responses=True)

def test_caching(redis_client): """Test caching logic.""" # Test cache miss assert redis_client.get("test_key") is None

# Test cache set
redis_client.setex("test_key", 60, "test_value")
assert redis_client.get("test_key") == "test_value"

# Test expiration
assert redis_client.ttl("test_key") <= 60

def test_session_management(redis_client): """Test session operations.""" session_manager = SessionManager(redis_client)

# Create session
session_id = session_manager.create_session(user_id=123)
assert session_id is not None

# Get session
session = session_manager.get_session(session_id)
assert session['user_id'] == 123

# Delete session
assert session_manager.delete_session(session_id) is True
assert session_manager.get_session(session_id) is None

Examples

Example 1: User Session Management with Redis

import redis import json import uuid from datetime import datetime, timedelta

class UserSessionManager: """Complete user session management with Redis."""

def __init__(self, redis_client: redis.Redis, ttl: int = 1800):
    self.redis = redis_client
    self.ttl = ttl

def create_session(self, user_id: int, user_data: dict = None) -> str:
    """Create new user session."""
    session_id = str(uuid.uuid4())
    session_key = f"session:{session_id}"

    session_data = {
        "user_id": user_id,
        "created_at": datetime.utcnow().isoformat(),
        "last_accessed": datetime.utcnow().isoformat(),
        "data": user_data or {}
    }

    # Store session with TTL
    self.redis.setex(session_key, self.ttl, json.dumps(session_data))

    # Track user's active sessions
    self.redis.sadd(f"user:{user_id}:sessions", session_id)

    return session_id

def get_session(self, session_id: str) -> dict:
    """Get session and refresh TTL."""
    session_key = f"session:{session_id}"
    session_data = self.redis.get(session_key)

    if session_data:
        session = json.loads(session_data)
        session['last_accessed'] = datetime.utcnow().isoformat()

        # Refresh TTL
        self.redis.setex(session_key, self.ttl, json.dumps(session))

        return session

    return None

def delete_session(self, session_id: str) -> bool:
    """Delete session."""
    session = self.get_session(session_id)
    if not session:
        return False

    user_id = session['user_id']

    # Remove session
    self.redis.delete(f"session:{session_id}")

    # Remove from user's session set
    self.redis.srem(f"user:{user_id}:sessions", session_id)

    return True

def delete_all_user_sessions(self, user_id: int):
    """Delete all sessions for a user."""
    sessions_key = f"user:{user_id}:sessions"
    session_ids = self.redis.smembers(sessions_key)

    for session_id in session_ids:
        self.redis.delete(f"session:{session_id}")

    self.redis.delete(sessions_key)

def get_user_sessions(self, user_id: int) -> list:
    """Get all active sessions for a user."""
    sessions_key = f"user:{user_id}:sessions"
    session_ids = self.redis.smembers(sessions_key)

    sessions = []
    for session_id in session_ids:
        session = self.get_session(session_id)
        if session:
            session['session_id'] = session_id
            sessions.append(session)

    return sessions

Usage

r = redis.Redis(decode_responses=True) session_mgr = UserSessionManager(r)

Create session

session_id = session_mgr.create_session( user_id=123, user_data={"role": "admin", "permissions": ["read", "write"]} )

Get session

session = session_mgr.get_session(session_id) print(f"User ID: {session['user_id']}")

List all user sessions

sessions = session_mgr.get_user_sessions(123) print(f"Active sessions: {len(sessions)}")

Logout (delete session)

session_mgr.delete_session(session_id)

Example 2: Real-Time Leaderboard

import redis import time

class Leaderboard: """Real-time leaderboard using Redis sorted sets."""

def __init__(self, redis_client: redis.Redis, name: str):
    self.redis = redis_client
    self.key = f"leaderboard:{name}"

def add_score(self, player_id: str, score: float):
    """Add or update player score."""
    self.redis.zadd(self.key, {player_id: score})

def increment_score(self, player_id: str, increment: float):
    """Increment player score."""
    self.redis.zincrby(self.key, increment, player_id)

def get_top(self, count: int = 10) -> list:
    """Get top players."""
    # ZREVRANGE for highest scores first
    players = self.redis.zrevrange(self.key, 0, count - 1, withscores=True)

    return [
        {
            "rank": idx + 1,
            "player_id": player_id,
            "score": score
        }
        for idx, (player_id, score) in enumerate(players)
    ]

def get_rank(self, player_id: str) -> dict:
    """Get player rank and score."""
    score = self.redis.zscore(self.key, player_id)
    if score is None:
        return None

    # ZREVRANK for rank (0-indexed, highest first)
    rank = self.redis.zrevrank(self.key, player_id)

    return {
        "player_id": player_id,
        "rank": rank + 1 if rank is not None else None,
        "score": score
    }

def get_around(self, player_id: str, count: int = 5) -> list:
    """Get players around a specific player."""
    rank = self.redis.zrevrank(self.key, player_id)
    if rank is None:
        return []

    # Get players before and after
    start = max(0, rank - count)
    end = rank + count

    players = self.redis.zrevrange(self.key, start, end, withscores=True)

    return [
        {
            "rank": start + idx + 1,
            "player_id": pid,
            "score": score,
            "is_current": pid == player_id
        }
        for idx, (pid, score) in enumerate(players)
    ]

def get_total_players(self) -> int:
    """Get total number of players."""
    return self.redis.zcard(self.key)

def remove_player(self, player_id: str) -> bool:
    """Remove player from leaderboard."""
    return self.redis.zrem(self.key, player_id) > 0

Usage

r = redis.Redis(decode_responses=True) leaderboard = Leaderboard(r, "global")

Add scores

leaderboard.add_score("alice", 1500) leaderboard.add_score("bob", 2000) leaderboard.add_score("charlie", 1800) leaderboard.increment_score("alice", 200) # alice now at 1700

Get top 10

top_players = leaderboard.get_top(10) for player in top_players: print(f"#{player['rank']}: {player['player_id']} - {player['score']}")

Get player rank

alice_stats = leaderboard.get_rank("alice") print(f"Alice is rank {alice_stats['rank']} with {alice_stats['score']} points")

Get players around alice

nearby = leaderboard.get_around("alice", count=2) for player in nearby: marker = " <-- YOU" if player['is_current'] else "" print(f"#{player['rank']}: {player['player_id']} - {player['score']}{marker}")

Example 3: Distributed Rate Limiter

import redis import time

class RateLimiter: """Distributed rate limiter using Redis."""

def __init__(self, redis_client: redis.Redis):
    self.redis = redis_client

    # Lua script for atomic rate limiting
    self.rate_limit_script = self.redis.register_script("""
        local key = KEYS[1]
        local limit = tonumber(ARGV[1])
        local window = tonumber(ARGV[2])

        local current = redis.call('INCR', key)

        if current == 1 then
            redis.call('EXPIRE', key, window)
        end

        if current > limit then
            return {0, limit, current - 1}
        else
            return {1, limit, current}
        end
    """)

def check_rate_limit(self, identifier: str, limit: int, window: int) -> dict:
    """
    Check if request is within rate limit.

    Args:
        identifier: User ID, IP address, or API key
        limit: Maximum requests allowed
        window: Time window in seconds

    Returns:
        dict with allowed (bool), limit, current, remaining
    """
    key = f"rate_limit:{identifier}:{int(time.time() // window)}"

    allowed, max_limit, current = self.rate_limit_script(
        keys=[key],
        args=[limit, window]
    )

    return {
        "allowed": bool(allowed),
        "limit": max_limit,
        "current": current,
        "remaining": max(0, max_limit - current),
        "reset_at": (int(time.time() // window) + 1) * window
    }

def sliding_window_check(self, identifier: str, limit: int, window: int) -> dict:
    """
    Sliding window rate limiter using sorted sets.
    More accurate but slightly more expensive.
    """
    key = f"rate_limit:sliding:{identifier}"
    now = time.time()
    window_start = now - window

    # Remove old entries
    self.redis.zremrangebyscore(key, 0, window_start)

    # Count current requests
    current = self.redis.zcard(key)

    if current < limit:
        # Add new request
        self.redis.zadd(key, {str(now): now})
        self.redis.expire(key, window)

        return {
            "allowed": True,
            "limit": limit,
            "current": current + 1,
            "remaining": limit - current - 1
        }
    else:
        return {
            "allowed": False,
            "limit": limit,
            "current": current,
            "remaining": 0
        }

Usage

r = redis.Redis(decode_responses=True) limiter = RateLimiter(r)

API rate limiting: 100 requests per minute

user_id = "user_123" result = limiter.check_rate_limit(user_id, limit=100, window=60)

if result["allowed"]: print(f"Request allowed. {result['remaining']} requests remaining.") # Process request else: print(f"Rate limit exceeded. Try again at {result['reset_at']}") # Return 429 Too Many Requests

More accurate sliding window

result = limiter.sliding_window_check(user_id, limit=100, window=60)

Example 4: Distributed Job Queue

import redis import json import time import uuid from typing import Optional, Callable

class JobQueue: """Distributed job queue with Redis."""

def __init__(self, redis_client: redis.Redis, queue_name: str = "default"):
    self.redis = redis_client
    self.queue_name = queue_name
    self.queue_key = f"queue:{queue_name}"
    self.processing_key = f"queue:{queue_name}:processing"

def enqueue(self, job_type: str, payload: dict, priority: int = 0) -> str:
    """
    Add job to queue.

    Args:
        job_type: Type of job (for routing to workers)
        payload: Job data
        priority: Higher priority = processed first (0 = normal)

    Returns:
        job_id
    """
    job_id = str(uuid.uuid4())

    job_data = {
        "id": job_id,
        "type": job_type,
        "payload": payload,
        "enqueued_at": time.time(),
        "attempts": 0
    }

    # Add to queue (use ZADD for priority queue)
    score = -priority  # Negative for higher priority first
    self.redis.zadd(self.queue_key, {json.dumps(job_data): score})

    return job_id

def dequeue(self, timeout: int = 0) -> Optional[dict]:
    """
    Get next job from queue.

    Args:
        timeout: Block for this many seconds (0 = no blocking)

    Returns:
        Job data or None
    """
    # Get highest priority job (lowest score)
    jobs = self.redis.zrange(self.queue_key, 0, 0)

    if not jobs:
        if timeout > 0:
            time.sleep(min(timeout, 1))
            return self.dequeue(timeout - 1)
        return None

    job_json = jobs[0]

    # Move to processing set atomically
    pipe = self.redis.pipeline()
    pipe.zrem(self.queue_key, job_json)
    pipe.zadd(self.processing_key, {job_json: time.time()})
    pipe.execute()

    job_data = json.loads(job_json)
    job_data['attempts'] += 1

    return job_data

def complete(self, job_data: dict):
    """Mark job as completed."""
    job_json = json.dumps({
        k: v for k, v in job_data.items()
        if k != 'attempts'
    })

    # Remove from processing
    self.redis.zrem(self.processing_key, job_json)

def retry(self, job_data: dict, delay: int = 0):
    """Retry failed job."""
    job_json = json.dumps({
        k: v for k, v in job_data.items()
        if k != 'attempts'
    })

    # Remove from processing
    self.redis.zrem(self.processing_key, job_json)

    # Re-enqueue with delay
    if delay > 0:
        time.sleep(delay)

    self.redis.zadd(self.queue_key, {job_json: 0})

def get_stats(self) -> dict:
    """Get queue statistics."""
    return {
        "queued": self.redis.zcard(self.queue_key),
        "processing": self.redis.zcard(self.processing_key)
    }

Worker example

class Worker: """Job worker."""

def __init__(self, queue: JobQueue, handlers: dict):
    self.queue = queue
    self.handlers = handlers

def process_jobs(self):
    """Process jobs from queue."""
    print("Worker started. Waiting for jobs...")

    while True:
        job = self.queue.dequeue(timeout=5)

        if job:
            print(f"Processing job {job['id']} (type: {job['type']})")

            try:
                # Get handler for job type
                handler = self.handlers.get(job['type'])

                if handler:
                    handler(job['payload'])
                    self.queue.complete(job)
                    print(f"Job {job['id']} completed")
                else:
                    print(f"No handler for job type: {job['type']}")
                    self.queue.complete(job)

            except Exception as e:
                print(f"Job {job['id']} failed: {e}")

                if job['attempts'] < 3:
                    # Retry with exponential backoff
                    delay = 2 ** job['attempts']
                    print(f"Retrying in {delay}s...")
                    self.queue.retry(job, delay=delay)
                else:
                    print(f"Job {job['id']} failed permanently")
                    self.queue.complete(job)

Usage

r = redis.Redis(decode_responses=True) queue = JobQueue(r, "email_queue")

Enqueue jobs

job_id = queue.enqueue("send_email", { "to": "user@example.com", "subject": "Welcome!", "body": "Thanks for signing up." }, priority=1)

Define handlers

def send_email_handler(payload): print(f"Sending email to {payload['to']}") # Email sending logic here time.sleep(1) # Simulate work

handlers = { "send_email": send_email_handler }

Start worker

worker = Worker(queue, handlers)

worker.process_jobs() # This blocks - run in separate process

Example 5: Real-Time Event Streaming

import redis import json import time from typing import Callable, Optional

class EventStream: """Real-time event streaming with Redis Streams."""

def __init__(self, redis_client: redis.Redis, stream_name: str):
    self.redis = redis_client
    self.stream_name = stream_name

def publish(self, event_type: str, data: dict) -> str:
    """Publish event to stream."""
    event = {
        "type": event_type,
        "data": json.dumps(data),
        "timestamp": time.time()
    }

    # Add to stream (returns auto-generated ID)
    event_id = self.redis.xadd(self.stream_name, event, maxlen=10000)
    return event_id

def read_events(self, last_id: str = '0', count: int = 10) -> list:
    """Read events from stream."""
    events = self.redis.xread(
        {self.stream_name: last_id},
        count=count,
        block=1000  # 1 second timeout
    )

    if not events:
        return []

    _, event_list = events[0]

    return [
        {
            "id": event_id,
            "type": event_data[b'type'].decode(),
            "data": json.loads(event_data[b'data'].decode()),
            "timestamp": float(event_data[b'timestamp'])
        }
        for event_id, event_data in event_list
    ]

def create_consumer_group(self, group_name: str):
    """Create consumer group for parallel processing."""
    try:
        self.redis.xgroup_create(
            name=self.stream_name,
            groupname=group_name,
            id='0',
            mkstream=True
        )
    except redis.ResponseError as e:
        if "BUSYGROUP" not in str(e):
            raise

def consume_events(self, group_name: str, consumer_name: str,
                  count: int = 10) -> list:
    """Consume events as part of consumer group."""
    events = self.redis.xreadgroup(
        groupname=group_name,
        consumername=consumer_name,
        streams={self.stream_name: '>'},
        count=count,
        block=5000
    )

    if not events:
        return []

    _, event_list = events[0]

    return [
        {
            "id": event_id,
            "type": event_data[b'type'].decode(),
            "data": json.loads(event_data[b'data'].decode()),
            "timestamp": float(event_data[b'timestamp'])
        }
        for event_id, event_data in event_list
    ]

def acknowledge(self, group_name: str, event_id: str):
    """Acknowledge processed event."""
    self.redis.xack(self.stream_name, group_name, event_id)

def get_pending(self, group_name: str) -> list:
    """Get pending (unacknowledged) events."""
    pending = self.redis.xpending_range(
        name=self.stream_name,
        groupname=group_name,
        min='-',
        max='+',
        count=100
    )

    return pending

Usage Example: Activity Feed

r = redis.Redis() activity_stream = EventStream(r, "user_activity")

Publish events

activity_stream.publish("user_signup", { "user_id": 123, "email": "alice@example.com" })

activity_stream.publish("post_created", { "user_id": 123, "post_id": 456, "title": "My First Post" })

Read events (simple consumer)

last_id = '0' while True: events = activity_stream.read_events(last_id, count=10)

for event in events:
    print(f"Event: {event['type']}")
    print(f"Data: {event['data']}")
    last_id = event['id']

if not events:
    break

Consumer group example

activity_stream.create_consumer_group("processors")

Worker consuming events

while True: events = activity_stream.consume_events( group_name="processors", consumer_name="worker-1", count=10 )

for event in events:
    try:
        # Process event
        process_event(event)

        # Acknowledge
        activity_stream.acknowledge("processors", event['id'])
    except Exception as e:
        print(f"Failed to process event {event['id']}: {e}")
        # Event remains unacknowledged for retry

Example 6: Cache-Aside Pattern with Multi-Level Caching

import redis import json import hashlib from typing import Optional, Any, Callable

class MultiLevelCache: """Multi-level caching with Redis and local cache."""

def __init__(self, redis_client: redis.Redis,
             local_cache_size: int = 100,
             local_ttl: int = 60,
             redis_ttl: int = 3600):
    self.redis = redis_client
    self.local_cache = {}
    self.local_cache_size = local_cache_size
    self.local_ttl = local_ttl
    self.redis_ttl = redis_ttl

def _make_key(self, namespace: str, key: str) -> str:
    """Generate cache key."""
    return f"cache:{namespace}:{key}"

def get(self, namespace: str, key: str,
        compute_fn: Optional[Callable] = None) -> Optional[Any]:
    """
    Get value from cache with fallback to compute function.

    Lookup order: Local cache → Redis → Compute function
    """
    cache_key = self._make_key(namespace, key)

    # Level 1: Local cache
    if cache_key in self.local_cache:
        entry = self.local_cache[cache_key]
        if time.time() < entry['expires_at']:
            return entry['value']
        else:
            del self.local_cache[cache_key]

    # Level 2: Redis cache
    redis_value = self.redis.get(cache_key)
    if redis_value:
        value = json.loads(redis_value)

        # Populate local cache
        self._set_local(cache_key, value)

        return value

    # Level 3: Compute function
    if compute_fn:
        value = compute_fn()
        if value is not None:
            self.set(namespace, key, value)
        return value

    return None

def set(self, namespace: str, key: str, value: Any):
    """Set value in both cache levels."""
    cache_key = self._make_key(namespace, key)
    serialized = json.dumps(value)

    # Set in Redis
    self.redis.setex(cache_key, self.redis_ttl, serialized)

    # Set in local cache
    self._set_local(cache_key, value)

def _set_local(self, key: str, value: Any):
    """Set value in local cache with LRU eviction."""
    # Simple LRU: remove oldest if at capacity
    if len(self.local_cache) >= self.local_cache_size:
        # Remove oldest entry
        oldest_key = min(
            self.local_cache.keys(),
            key=lambda k: self.local_cache[k]['expires_at']
        )
        del self.local_cache[oldest_key]

    self.local_cache[key] = {
        'value': value,
        'expires_at': time.time() + self.local_ttl
    }

def delete(self, namespace: str, key: str):
    """Delete from all cache levels."""
    cache_key = self._make_key(namespace, key)

    # Delete from Redis
    self.redis.delete(cache_key)

    # Delete from local cache
    if cache_key in self.local_cache:
        del self.local_cache[cache_key]

def invalidate_namespace(self, namespace: str):
    """Invalidate all keys in namespace."""
    pattern = f"cache:{namespace}:*"

    # Delete from Redis
    for key in self.redis.scan_iter(match=pattern, count=100):
        self.redis.delete(key)

    # Delete from local cache
    to_delete = [
        k for k in self.local_cache.keys()
        if k.startswith(f"cache:{namespace}:")
    ]
    for k in to_delete:
        del self.local_cache[k]

Usage

r = redis.Redis(decode_responses=True) cache = MultiLevelCache(r)

def get_user(user_id: int) -> dict: """Get user with multi-level caching.""" return cache.get( namespace="users", key=str(user_id), compute_fn=lambda: database.query_user(user_id) )

First call: Queries database, caches result

user = get_user(123)

Second call: Returns from local cache (fastest)

user = get_user(123)

Update user

def update_user(user_id: int, data: dict): database.update_user(user_id, data)

# Invalidate cache
cache.delete("users", str(user_id))

Invalidate all user caches

cache.invalidate_namespace("users")

Example 7: Geo-Location with Redis

import redis

class GeoLocation: """Geo-spatial indexing and queries with Redis."""

def __init__(self, redis_client: redis.Redis, index_name: str):
    self.redis = redis_client
    self.key = f"geo:{index_name}"

def add_location(self, location_id: str, longitude: float, latitude: float):
    """Add location to geo index."""
    self.redis.geoadd(self.key, longitude, latitude, location_id)

def add_locations(self, locations: list):
    """Batch add locations.

    Args:
        locations: List of (location_id, longitude, latitude) tuples
    """
    self.redis.geoadd(self.key, *[
        item for loc in locations
        for item in (loc[1], loc[2], loc[0])
    ])

def get_position(self, location_id: str) -> tuple:
    """Get coordinates of a location."""
    result = self.redis.geopos(self.key, location_id)
    if result and result[0]:
        return result[0]  # (longitude, latitude)
    return None

def find_nearby(self, longitude: float, latitude: float,
               radius: float, unit: str = 'km', count: int = None) -> list:
    """
    Find locations within radius.

    Args:
        longitude: Center longitude
        latitude: Center latitude
        radius: Search radius
        unit: Distance unit ('m', 'km', 'mi', 'ft')
        count: Maximum results
    """
    args = {
        'longitude': longitude,
        'latitude': latitude,
        'radius': radius,
        'unit': unit,
        'withdist': True,
        'withcoord': True,
        'sort': 'ASC'
    }

    if count:
        args['count'] = count

    results = self.redis.georadius(self.key, **args)

    return [
        {
            'location_id': location_id,
            'distance': distance,
            'coordinates': (longitude, latitude)
        }
        for location_id, distance, (longitude, latitude) in results
    ]

def find_nearby_member(self, location_id: str, radius: float,
                      unit: str = 'km', count: int = None) -> list:
    """Find locations near an existing member."""
    args = {
        'member': location_id,
        'radius': radius,
        'unit': unit,
        'withdist': True,
        'sort': 'ASC'
    }

    if count:
        args['count'] = count

    results = self.redis.georadiusbymember(self.key, **args)

    return [
        {
            'location_id': loc_id,
            'distance': distance
        }
        for loc_id, distance in results
        if loc_id != location_id  # Exclude self
    ]

def distance_between(self, location_id1: str, location_id2: str,
                    unit: str = 'km') -> float:
    """Calculate distance between two locations."""
    return self.redis.geodist(self.key, location_id1, location_id2, unit)

Usage Example: Restaurant finder

r = redis.Redis(decode_responses=True) restaurants = GeoLocation(r, "restaurants")

Add restaurants

restaurants.add_locations([ ("rest1", -122.4194, 37.7749), # San Francisco ("rest2", -122.4068, 37.7849), ("rest3", -122.4312, 37.7652), ])

Find restaurants near coordinates

nearby = restaurants.find_nearby( longitude=-122.4194, latitude=37.7749, radius=5, unit='km', count=10 )

for restaurant in nearby: print(f"{restaurant['location_id']}: {restaurant['distance']:.2f} km away")

Find restaurants near a specific restaurant

similar = restaurants.find_nearby_member("rest1", radius=2, unit='km')

Get distance between two restaurants

distance = restaurants.distance_between("rest1", "rest2", unit='km') print(f"Distance: {distance:.2f} km")

Quick Reference

Common Operations

Connection

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

Strings

r.set('key', 'value') r.setex('key', 3600, 'value') # With TTL r.get('key') r.incr('counter')

Hashes

r.hset('user:123', 'name', 'Alice') r.hset('user:123', mapping={'name': 'Alice', 'age': 30}) r.hget('user:123', 'name') r.hgetall('user:123')

Lists

r.lpush('queue', 'item') r.rpush('queue', 'item') r.lpop('queue') r.lrange('queue', 0, -1)

Sets

r.sadd('tags', 'python', 'redis') r.smembers('tags') r.sismember('tags', 'python')

Sorted Sets

r.zadd('leaderboard', {'alice': 100, 'bob': 200}) r.zrange('leaderboard', 0, -1, withscores=True) r.zrank('leaderboard', 'alice')

Expiration

r.expire('key', 3600) r.ttl('key')

Pipelining

pipe = r.pipeline() pipe.set('key1', 'value1') pipe.set('key2', 'value2') results = pipe.execute()

Time Complexity

• GET, SET: O(1)

• HGET, HSET: O(1)

• LPUSH, RPUSH, LPOP, RPOP: O(1)

• SADD, SREM, SISMEMBER: O(1)

• ZADD, ZREM: O(log(N))

• ZRANGE, ZREVRANGE: O(log(N)+M) where M is result size

• SCAN, SSCAN, HSCAN, ZSCAN: O(1) per iteration

Skill Version: 1.0.0 Last Updated: October 2025 Skill Category: State Management, Distributed Systems, Performance Optimization Compatible With: redis-py, Redis 6.0+, Redis 7.0+