Installer avec Codex ou Claude Copiez ce prompt, collez-le dans Codex, Claude ou un autre assistant, puis laissez-le vérifier la page du skill et l'installer pour vous.
Une commande directe contourne le prompt de vérification. Examinez la source avant de l'exécuter.
Redis data structure patterns, caching strategies, distributed locks, rate limiting, pub/sub, and connection management for production applications.
origin
ECC
Redis Patterns
Quick reference for Redis best practices across common backend use cases.
How It Works
Redis is an in-memory data structure store that supports strings, hashes, lists, sets, sorted sets, streams, and more. Individual Redis commands are atomic on a single instance; multi-step workflows require Lua scripts, MULTI/EXEC transactions, or explicit synchronization to stay atomic. Data is optionally persisted via RDB snapshots or AOF logs. Clients communicate over TCP using the RESP protocol; connection pools are essential to avoid per-request handshake overhead.
All primary examples use redis.asyncio — the async-native client bundled with redis-py ≥ 4.2. Use it for any FastAPI / asyncio application. A short sync note is included at the end of Connection Management for scripts and CLI utilities.
When to Activate
Adding caching to an application
Implementing rate limiting or throttling
Building distributed locks or coordination
Setting up session or token storage
Using Pub/Sub or Redis Streams for messaging
Configuring Redis in production (pooling, eviction, clustering)
Data Structure Cheat Sheet
Use Case
Structure
Example Key
Simple cache
String
product:123
User session
Hash
session:abc
Leaderboard
Sorted Set
scores:weekly
Unique visitors
Set
visitors:2024-01-01
Activity feed
List
feed:user:456
Event stream
Stream
events:orders
Counters / rate limits
String (INCR)
ratelimit:user:123
Bloom filter / HLL
HyperLogLog
hll:pageviews
Core Patterns
Cache-Aside (Lazy Loading)
import json
import redis.asyncio as redis
# Module-level pool — create once at app startup, reuse everywhere.
pool = redis.ConnectionPool.from_url(
"redis://localhost:6379/0",
decode_responses=True,
max_connections=20,
)
r = redis.Redis(connection_pool=pool)
asyncdefget_product(product_id: int) -> dict:
cache_key = f"product:{product_id}"
cached = await r.get(cache_key)
if cached:
return json.loads(cached)
product = await db.fetchrow("SELECT * FROM products WHERE id = $1", product_id)
await r.setex(cache_key, 3600, json.dumps(product)) # TTL: 1 hourreturn product
FastAPI lifespan wiring — close the pool cleanly on shutdown:
from contextlib import asynccontextmanager
from fastapi import FastAPI
import redis.asyncio as redis
pool: redis.ConnectionPool | None = None@asynccontextmanagerasyncdeflifespan(app: FastAPI):
global pool
pool = redis.ConnectionPool.from_url(
"redis://localhost:6379/0",
decode_responses=True,
max_connections=20,
socket_connect_timeout=2,
socket_timeout=2,
)
app.state.redis = redis.Redis(connection_pool=pool)
yieldawait pool.aclose()
app = FastAPI(lifespan=lifespan)
Write-Through Cache
asyncdefupdate_product(product_id: int, data: dict) -> None:
# Write to DB firstawait db.execute("UPDATE products SET ... WHERE id = $1", product_id)
# Immediately update cache
cache_key = f"product:{product_id}"await r.setex(cache_key, 3600, json.dumps(data))
Cache Invalidation
# Tag-based invalidation — group related keys under a setasyncdefcache_product(product_id: int, category_id: int, data: dict) -> None:
key = f"product:{product_id}"
tag = f"tag:category:{category_id}"asyncwith r.pipeline(transaction=True) as pipe:
await pipe.setex(key, 3600, json.dumps(data))
await pipe.sadd(tag, key)
await pipe.expire(tag, 3600)
await pipe.execute()
asyncdefinvalidate_category(category_id: int) -> None:
tag = f"tag:category:{category_id}"
keys = await r.smembers(tag)
if keys:
await r.delete(*keys)
await r.delete(tag)
Session Storage
import time
import uuid
asyncdefcreate_session(user_id: int, ttl: int = 86400) -> str:
session_id = str(uuid.uuid4())
key = f"session:{session_id}"asyncwith r.pipeline(transaction=True) as pipe:
await pipe.hset(key, mapping={
"user_id": user_id,
"created_at": int(time.time()),
})
await pipe.expire(key, ttl)
await pipe.execute()
return session_id
asyncdefget_session(session_id: str) -> dict | None:
data = await r.hgetall(f"session:{session_id}")
return data if data elseNoneasyncdefdelete_session(session_id: str) -> None:
await r.delete(f"session:{session_id}")
Always set a TTL. Keys without TTL accumulate indefinitely and cause memory pressure.
Connection Management
Async Connection Pool (FastAPI / asyncio — default)
import redis.asyncio as redis
pool = redis.ConnectionPool.from_url(
"redis://localhost:6379/0",
decode_responses=True,
max_connections=20,
socket_connect_timeout=2,
socket_timeout=2,
)
r = redis.Redis(connection_pool=pool)
# Shared pool across multiple client handles (e.g., separate read/write clients)
r_read = redis.Redis(connection_pool=pool)
r_write = redis.Redis(connection_pool=pool)
# Shutdown (call from lifespan teardown)await pool.aclose()
Sync (scripts / CLI only)
For one-off scripts or management CLIs that do not run inside an asyncio event loop:
import redis # sync client — do NOT use in FastAPI request handlers
pool = redis.ConnectionPool(
host="localhost", port=6379, db=0,
max_connections=5, decode_responses=True,
)
r = redis.Redis(connection_pool=pool)
Cluster Mode
from redis.asyncio.cluster import RedisCluster
r = RedisCluster.from_url(
"redis://redis-1:6379",
decode_responses=True,
skip_full_coverage_check=True,
)
For multi-process deployments, replace the in-process asyncio.Lock with acquire_lock/release_lock from the Distributed Locks section above.
Examples
Add caching to a FastAPI endpoint:
Use cache-aside with await r.setex(...) and a 5-minute TTL. Key on the request parameters. Wire the pool via app.state.redis in the lifespan context (see FastAPI wiring above).
Rate-limit an API by user:
Use fixed-window with async with r.pipeline(transaction=True) for low-traffic endpoints; use sliding-window Lua for accurate per-user throttling.
Coordinate a background job across workers:
Use await acquire_lock(...) with a TTL that exceeds the expected job duration. Always release in a finally block.
Fan-out notifications to multiple subscribers:
Use Pub/Sub (async with r.pubsub()) for fire-and-forget. Switch to Streams if you need guaranteed delivery or replay for late consumers.
Quick Reference
Pattern
When to Use
Cache-aside
Read-heavy, tolerate slight staleness
Write-through
Strong consistency required
Distributed lock
Prevent concurrent access to a resource
Sliding window rate limit
Accurate per-user throttling
Redis Streams
Durable event queue with consumer groups
Pub/Sub
Broadcast with no delivery guarantees needed
Sorted Set leaderboard
Ranked scoring, pagination
HyperLogLog
Approximate unique count at low memory
Related
Skill: postgres-patterns — relational data patterns
Skill: backend-patterns — API and service layer patterns
