Redis Patterns

Production Redis patterns for Python async applications. Caching, pub/sub, streams, and distributed operations.

When to Apply

• Implementing or modifying cache logic
• Working with Redis context (market data, LLM context)
• Building rate limiting
• Implementing distributed locks
• Event processing with streams

1. Cache-Aside Pattern (CRITICAL)

import json
import redis.asyncio as aioredis

redis = aioredis.from_url("redis://localhost:6379", decode_responses=True)

async def cache_aside(key: str, ttl: int, fetcher) -> dict:
    """Get from cache or fetch and cache."""
    cached = await redis.get(key)
    if cached:
        return json.loads(cached)
    
    value = await fetcher()
    await redis.set(key, json.dumps(value, default=str), ex=ttl)
    return value

# Usage
data = await cache_aside(
    "market:context:btc",
    ttl=300,
    fetcher=lambda: build_btc_context()
)

Cache invalidation

async def invalidate_pattern(pattern: str):
    """Invalidate all keys matching pattern using SCAN."""
    cursor = 0
    while True:
        cursor, keys = await redis.scan(cursor, match=pattern, count=100)
        if keys:
            await redis.delete(*keys)
        if cursor == 0:
            break

# NEVER use KEYS * in production - blocks Redis
# ALWAYS use SCAN for pattern matching
await invalidate_pattern("market:context:*")

2. Rate Limiting - Sliding Window (CRITICAL)

import time
import uuid

async def is_rate_limited(client_id: str, limit: int, window_seconds: int) -> bool:
    key = f"ratelimit:{client_id}"
    now = time.time()
    window_start = now - window_seconds
    
    pipe = redis.pipeline()
    pipe.zremrangebyscore(key, 0, window_start)
    pipe.zadd(key, {f"{now}:{uuid.uuid4().hex[:8]}": now})
    pipe.zcard(key)
    pipe.expire(key, window_seconds)
    
    results = await pipe.execute()
    count = results[2]
    return count > limit

# Usage in FastAPI
@router.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    if await is_rate_limited(request.user_id, limit=10, window_seconds=60):
        raise HTTPException(429, "Rate limit exceeded")
    # ... process

3. Pub/Sub (HIGH)

async def publish_event(channel: str, event: dict):
    await redis.publish(channel, json.dumps(event, default=str))

async def subscribe_events(channels: list[str], handler):
    pubsub = redis.pubsub()
    await pubsub.subscribe(*channels)
    
    async for message in pubsub.listen():
        if message["type"] == "message":
            event = json.loads(message["data"])
            await handler(message["channel"], event)

# Usage
await publish_event("market:alerts", {
    "type": "whale_transfer",
    "symbol": "BTC",
    "volume_usd": 50_000_000
})

4. Streams for Event Processing (HIGH)

async def produce_event(stream: str, event: dict):
    await redis.xadd(stream, event, maxlen=10000)

async def consume_stream(stream: str, group: str, consumer: str, handler):
    # Create group if not exists
    try:
        await redis.xgroup_create(stream, group, "0", mkstream=True)
    except Exception:
        pass  # Group exists
    
    while True:
        results = await redis.xreadgroup(
            group, consumer,
            {stream: ">"},
            count=10,
            block=5000
        )
        
        if not results:
            continue
        
        for stream_name, messages in results:
            for msg_id, fields in messages:
                await handler(fields)
                await redis.xack(stream_name, group, msg_id)

Use streams over pub/sub when you need:

• Message durability (pub/sub loses messages if no subscriber)
• Consumer groups (load balancing)
• Message acknowledgment

5. Distributed Lock (HIGH)

async def acquire_lock(resource: str, ttl: int = 30) -> str | None:
    token = uuid.uuid4().hex
    acquired = await redis.set(
        f"lock:{resource}", token, nx=True, ex=ttl
    )
    return token if acquired else None

async def release_lock(resource: str, token: str) -> bool:
    script = """
    if redis.call("GET", KEYS[1]) == ARGV[1] then
        return redis.call("DEL", KEYS[1])
    end
    return 0
    """
    result = await redis.eval(script, 1, f"lock:{resource}", token)
    return bool(result)

# Usage
token = await acquire_lock("update:btc_context")
if token:
    try:
        await update_btc_context()
    finally:
        await release_lock("update:btc_context", token)

6. Key Naming Convention

entity:id:field           - Single value
entity:id:subentity       - Hash/nested
ratelimit:client_id       - Rate limiting
lock:resource_name        - Distributed locks
stream:event_type         - Streams
cache:query_hash          - Query cache

# Our project patterns:
market:context:btc        - BTC market context
market:prices:latest      - Latest prices
news:search:query_hash    - News search cache
chat:session:user_id      - Chat session
scheduler:status          - Scheduler state

7. Anti-Patterns

• KEYS * in production (blocks Redis, use SCAN)
• No TTL on cache entries (memory grows unbounded)
• Large values >100KB without compression
• Pub/sub for durable messaging (use Streams)
• Not using pipelines for batch operations
• Storing sensitive data without encryption

8. Checklist

• All keys follow naming convention entity:id:field
• All cache entries have TTL
• SCAN used instead of KEYS for pattern matching
• Pipelines used for multiple sequential commands
• Streams used when durability is needed (not pub/sub)
• Connection pooling configured
• Distributed locks have TTL and proper release
• maxmemory-policy set (allkeys-lru recommended)