Redis v7IntermediateRedis
Redis data types, caching patterns, pub/sub, streams, and common use cases.
Data Types
1 topic
String, Hash, List, Set
# STRING — cache, counters, sessions
SET user:1:name "Alice"
GET user:1:name
SETEX session:abc 3600 "user_data" # Expire in 1 hour
INCR page:views # Atomic counter
# HASH — objects
HSET user:1 name Alice age 30 email [email protected]
HGET user:1 name
HGETALL user:1
HDEL user:1 email
# LIST — queues, feeds
LPUSH notifications:1 "New message"
RPUSH queue:jobs '{"type":"email"}'
LPOP queue:jobs
LRANGE notifications:1 0 9 # First 10
# SET — unique members, tags
SADD tags:post:1 python redis backend
SMEMBERS tags:post:1
SISMEMBER tags:post:1 python # 1 (exists)
SINTER tags:post:1 tags:post:2 # Common tags
# SORTED SET — leaderboards, rate limiting
ZADD leaderboard 1500 user:1
ZADD leaderboard 2300 user:2
ZRANGE leaderboard 0 9 WITHSCORES REV💡 Always set TTL on cache keys — SETEX or EXPIRE to prevent memory bloat
⚡ INCR is atomic — safe for counters without locking
Core Setup and Connectivity
2 topics
Run Redis Locally with Docker
Start a local Redis instance quickly for development and testing, exposing default port 6379.
# Start Redis container
docker run -d --name redis-dev -p 6379:6379 redis:7
# Verify server is up
docker exec -it redis-dev redis-cli ping
# PONG
# View logs if troubleshooting
docker logs redis-devUse a named container so you can restart it with docker start redis-dev.
Pin a version tag (e.g., redis:7.2) to avoid unexpected upgrades.
Connect from Python (redis-py)
Create a resilient Redis client with connection checks and basic error handling for app startup.
import redis
client = redis.Redis(
host='localhost',
port=6379,
db=0,
decode_responses=True,
socket_timeout=2,
socket_connect_timeout=2,
)
try:
print('Ping:', client.ping())
except redis.RedisError as e:
print('Redis connection failed:', e)
client.set('service:health', 'ok', ex=30)
print(client.get('service:health'))Set decode_responses=True to work with strings instead of bytes in Python.
Use short timeouts to fail fast during outages.
Caching Patterns (What, Why, How)
3 topics
Cache-Aside for Expensive API Calls
First check Redis; on miss fetch from source, store with TTL, then return. Great for read-heavy workloads.
import json
import redis
import requests
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
def get_user_profile(user_id: str):
key = f'user:profile:{user_id}'
cached = r.get(key)
if cached:
return json.loads(cached)
# Simulate expensive source call
data = requests.get(f'https://api.example.com/users/{user_id}', timeout=3).json()
# Cache for 5 minutes
r.set(key, json.dumps(data), ex=300)
return dataUse TTLs to prevent stale data from living forever.
Namespace keys (user:profile:123) for easier debugging and bulk invalidation.
Prevent Cache Stampede with Locking
Use a short-lived lock so only one worker refreshes missing data while others wait or serve fallback.
import time
import redis
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
def get_report(report_id: str):
data_key = f'report:{report_id}'
lock_key = f'lock:report:{report_id}'
data = r.get(data_key)
if data:
return data
# Try acquiring distributed lock (set if not exists)
if r.set(lock_key, '1', nx=True, ex=10):
try:
# Build expensive report
report = f'computed-report-{report_id}'
r.set(data_key, report, ex=120)
return report
finally:
r.delete(lock_key)
else:
# Another worker is building; short wait and retry once
time.sleep(0.2)
return r.get(data_key) or 'fallback-report'Always set lock expiration (EX) to avoid deadlocks.
For critical distributed locking semantics, evaluate Redlock carefully.
Invalidate by Pattern Safely
Use SCAN in production to iterate keys without blocking Redis, instead of KEYS.
# Dangerous in production (blocks on large datasets):
# redis-cli KEYS "user:profile:*" | xargs redis-cli DEL
# Safer approach with SCAN cursor iteration:
redis-cli --scan --pattern 'user:profile:*' | xargs -L 100 redis-cli DELPrefer event-driven invalidation when data changes over broad deletes.
Never use KEYS on large production datasets.
Patterns
1 topic
Caching & Rate Limiting
import redis
import json
from functools import wraps
r = redis.Redis(host="localhost", port=6379, decode_responses=True)
# Cache-aside pattern
def get_user(user_id: int):
key = f"user:{user_id}"
cached = r.get(key)
if cached:
return json.loads(cached)
user = db.get_user(user_id) # Fetch from DB
r.setex(key, 300, json.dumps(user)) # Cache 5 min
return user
# Rate limiting (sliding window)
def is_rate_limited(user_id: str, limit=10, window=60) -> bool:
key = f"rate:{user_id}"
pipeline = r.pipeline()
pipeline.incr(key)
pipeline.expire(key, window)
count, _ = pipeline.execute()
return count > limit
# Distributed lock
def with_lock(key: str, ttl=30):
acquired = r.set(f"lock:{key}", 1, nx=True, ex=ttl)
return bool(acquired)⚡ Use pipelines to batch Redis commands — reduces round trips
💡 nx=True in SET creates the key only if it doesn't exist — perfect for locks
Data Structures for Real Workloads
3 topics
Rate Limiting with INCR + EXPIRE
Implement fixed-window throttling (e.g., 100 requests/minute per user) using atomic counters.
import time
import redis
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
def allow_request(user_id: str, limit=100):
window = int(time.time() // 60)
key = f'rl:{user_id}:{window}'
count = r.incr(key)
if count == 1:
r.expire(key, 61)
return count <= limit
print(allow_request('u123'))For smoother control, consider sliding window or token bucket algorithms.
Keep expiration slightly longer than window boundary.
Leaderboards with Sorted Sets
Store scores in ZSET and fetch top users efficiently for gaming, sales dashboards, and rankings.
import redis
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
key = 'leaderboard:weekly'
# Update scores
r.zincrby(key, 50, 'alice')
r.zincrby(key, 20, 'bob')
r.zincrby(key, 80, 'carol')
# Top 3 descending
print(r.zrevrange(key, 0, 2, withscores=True))
# Get rank (0-based, descending)
print('alice rank:', r.zrevrank(key, 'alice'))Use periodic snapshots if historical leaderboard archives are needed.
Store metadata separately (e.g., user profile hash) and join in app layer.
Queues with Redis Streams
Use streams and consumer groups for durable event processing with acknowledgment and replay support.
import redis
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
stream = 'orders'
group = 'order-workers'
# Create group once
try:
r.xgroup_create(stream, group, id='0', mkstream=True)
except redis.ResponseError:
pass
# Producer
r.xadd(stream, {'order_id': 'A1001', 'amount': '42.50'})
# Consumer
messages = r.xreadgroup(group, 'consumer-1', {stream: '>'}, count=10, block=1000)
for _, entries in messages:
for msg_id, payload in entries:
print('Processing', msg_id, payload)
r.xack(stream, group, msg_id)Monitor pending entries list (XPENDING) to detect stuck consumers.
Use one consumer name per process instance.
Pub/Sub and Realtime Messaging
2 topics
Publish Notifications
Send lightweight, fire-and-forget events to subscribers (e.g., chat typing indicators or cache invalidation hints).
import redis
import json
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
payload = {'event': 'order.created', 'order_id': 'A1001'}
r.publish('events:orders', json.dumps(payload))Pub/Sub is not durable—messages are lost if subscribers are offline.
Use Streams when delivery guarantees are required.
Subscribe to Channels
Consume events in real time from one or more channels, common for websocket fan-out services.
import redis
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
pubsub = r.pubsub()
pubsub.subscribe('events:orders')
for message in pubsub.listen():
if message['type'] == 'message':
print('Received:', message['data'])Run subscriber loops in dedicated workers/threads.
Add reconnect logic for network interruptions.
Production Operations and Best Practices
3 topics
Persistence Choices (RDB vs AOF)
Choose snapshotting (RDB) for fast restarts and lower overhead, AOF for stronger durability, or both.
# redis.conf
# RDB snapshot every 5 min if at least 100 writes happened
save 300 100
# Append Only File for better durability
appendonly yes
appendfsync everysec
# Optional rewrite threshold
auto-aof-rewrite-percentage 100
auto-aof-rewrite-min-size 64mbUse both RDB + AOF in many production setups for balanced recovery and durability.
Test recovery procedures regularly, not only backups.
Security Baseline
Protect Redis by binding private interfaces, enabling ACL/auth, and using TLS where possible.
# redis.conf
bind 127.0.0.1
protected-mode yes
port 6379
# ACL example (Redis 6+)
user default off
user appuser on >S3cureP@ss ~app:* +@read +@write -@dangerousNever expose Redis directly to the public internet without strict controls.
Rotate credentials and separate users by least privilege.
Monitoring and Slow Query Troubleshooting
Track latency, memory, evictions, and expensive commands to catch performance issues early.
# Core health metrics
redis-cli INFO memory
redis-cli INFO stats
redis-cli INFO commandstats
# Check latency and slow operations
redis-cli LATENCY DOCTOR
redis-cli SLOWLOG GET 20
# Real-time command monitor (use carefully in prod)
redis-cli MONITORSet maxmemory and a proper eviction policy for cache workloads.
Alert on evicted_keys, rejected_connections, and used_memory_peak.
Related Cheat Sheets
More hands-on references connected to this topic.
PostgreSQLPostgreSQL cheat sheet: psql CLI, users & roles, backup & restore, VACUUM, locks, JSONB, CTEs, and indexes.
shared topics, same difficulty
Cloudflaresame difficulty, keyword overlap
FastAPIFastAPI routes, request models, dependency injection, authentication, and async patterns.
same difficulty, keyword overlap
KafkaPractical guide to building, operating, and scaling real-world Apache Kafka event streaming systems.
same difficulty, keyword overlap
Related Articles
Background reading and deeper explanations for this sheet.
Agent Routing vs Model Routing: Real Production Patterns, Tradeoffs, and Implementation Guide
Choosing between agent routing and model routing can make or break your AI product’s cost, speed, and reliability. In this practical guide, you’ll learn what each pattern is, when to use it, and how production teams combine both to ship robust systems. We’ll walk through real examples, architecture patterns, and code you can adapt immediately.
keyword overlap
Build an Agentic App with FastAPI and Azure AI Foundry: A Practical Beginner-to-Pro Guide
Learn how to design, build, and deploy an agentic application using FastAPI and Azure AI Foundry with practical, real-world examples. This guide walks you from architecture and setup to tool-calling, memory, observability, and production deployment patterns.
keyword overlap
Event-Driven Agents with Kafka Streams: A Practical Guide for Building Real-Time AI Workflows
Event-driven agents let you move from slow, request/response automation to responsive systems that react in milliseconds to real-world signals. In this hands-on guide, you’ll learn how to design, build, and operate agentic workflows using Apache Kafka and Kafka Streams, with practical patterns, code snippets, and production-ready advice.
keyword overlap
GenAI SaaS Architecture: A Practical Blueprint for Building, Scaling, and Securing AI Products
Designing a SaaS product on top of GenAI is more than calling an LLM API—it requires thoughtful architecture across product, data, safety, and operations. This practical guide walks you through a beginner-friendly yet deep system blueprint, with real-world patterns, code snippets, and deployment strategies you can use immediately.
keyword overlap