redis-patterns

name	redis-patterns
description	Redis data structure patterns, caching strategies, distributed locks, rate limiting, pub/sub, and connection management for production applications. Use when adding or debugging caching, a distributed lock, a rate limiter, session/token storage, or a Streams/Pub-Sub flow against the devskyy Redis. Do NOT use for durable relational data or transactional integrity (postgres-patterns — Redis is not a system of record), for schema/versioned state (database-migrations), or for the container's health/networking (docker-patterns).
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 use

• 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)

When NOT to use:

• Data that must survive a flush or a node loss → Postgres. Redis here runs maxmemory-policy noeviction with maxmemory 0 (verified below): unbounded, and nothing evicts to make room.
• Schema/versioned state → database-migrations. Relational modelling and indexes → postgres-patterns.
• "Redis container won't start / can't connect from the app" → docker-patterns first; this skill assumes a reachable, authenticated client.
• Any FLUSHALL/FLUSHDB against a shared instance — irreversible, STOP-AND-SHOW.

Inputs

Absent input = STOP. A cache key written to the wrong instance is invisible until it corrupts something.

1. A reachable, authenticated client. This instance requires auth: a bare redis-cli ping returns NOAUTH Authentication required. (observed 2026-07-28 [repro]). Use REDISCLI_AUTH="$REDIS_PASSWORD" (env, not argv). No password → stop; do not disable auth.
2. The target instance and DB index. Cache, queue, and rate-limit state should not share a logical DB by accident. Confirm with INFO server / dbsize before writing.
3. A TTL decision for every key you create. There is no eviction fallback here (see the maxmemory-policy check) — a key without a TTL is a permanent memory leak. If the TTL is genuinely unbounded, that key belongs in Postgres.
4. For any lock: an owner token and a TTL that exceeds the worst-case job duration. A lock released by a different holder is a correctness bug, which is why release must be the compare-and-delete Lua script, never a bare DEL.

Procedure

1. Confirm connectivity and identity (Verification check 1) before writing any code.
2. Pick the structure from the cheat sheet by access pattern, then design the key name (namespace:resource:id) and its TTL together — never the key alone.
3. Implement with redis.asyncio (the sync client blocks the FastAPI event loop) and a module-level pool created once at startup; wire teardown through the lifespan handler.
4. Make every multi-step sequence atomic: pipeline with transaction=True for grouped writes, or a Lua script when a read must gate a write (sliding-window limiter, lock release).
5. Set the TTL in the same call that writes the value (setex, or hset + expire inside one pipeline) — a crash between write and expire is exactly how untracked keys are born.
6. Verify TTL coverage and lock semantics against the real instance (Verification checks 2-3).
7. On failure paths, confirm the app degrades to the source of truth rather than serving stale or erroring: a cache outage must not be an application outage.

Verification

Each check names its command and pass condition. A command that returns an auth error or connects to the wrong instance is a dead gate, not a pass (bug-230). Results below are [repro] against the local dev container — they are not claims about any production Redis.

docker exec devskyy-redis sh -c 'REDISCLI_AUTH="$REDIS_PASSWORD" redis-cli ping; \
  REDISCLI_AUTH="$REDIS_PASSWORD" redis-cli dbsize; \
  REDISCLI_AUTH="$REDIS_PASSWORD" redis-cli info memory | grep -E "used_memory_human|maxmemory_human"'

PASS: PONG, plus a dbsize and memory figure you can compare against later. Observed 2026-07-28: PONG, dbsize 0, used_memory_human:1.23M, maxmemory_human:0B [repro]. maxmemory 0 = no cap, so TTLs are the only thing bounding memory on this instance.

docker exec devskyy-redis sh -c 'REDISCLI_AUTH="$REDIS_PASSWORD" redis-cli config get maxmemory-policy'

PASS: you have read the value and your key design matches it. Observed 2026-07-28: noeviction [repro] — under a memory cap this instance would error on writes rather than evict, so "the cache will just evict old keys" is false here. Any pattern in this skill that assumes allkeys-lru must be adjusted or the policy changed deliberately.

docker exec devskyy-redis sh -c '
  REDISCLI_AUTH="$REDIS_PASSWORD" redis-cli set lock:demo tok-1 PX 5000 NX
  REDISCLI_AUTH="$REDIS_PASSWORD" redis-cli set lock:demo tok-2 PX 5000 NX
  REDISCLI_AUTH="$REDIS_PASSWORD" redis-cli pttl lock:demo
  REDISCLI_AUTH="$REDIS_PASSWORD" redis-cli del lock:demo'

PASS: first SET returns OK, the second returns empty (nil — contended, correctly refused), pttl is a positive number ≤ 5000, and del returns 1. Observed 2026-07-28: OK, (empty), 4985, 1 [repro]. If the second SET also returns OK, NX is not being applied and your lock grants concurrent holders.

docker exec devskyy-redis sh -c 'REDISCLI_AUTH="$REDIS_PASSWORD" redis-cli --scan --count 1000 \
  | while read -r k; do t=$(REDISCLI_AUTH="$REDIS_PASSWORD" redis-cli ttl "$k"); \
    [ "$t" = "-1" ] && echo "NO-TTL $k"; done'

PASS: no NO-TTL lines. Every key that is not a deliberate, documented long-lived structure must carry a TTL — with noeviction and no maxmemory, a TTL-less key never leaves. [repro]

Prove the lock check can fail (rule 3): drop the NX flag from the second SET and re-run — it must return OK, showing the check distinguishes "lock held" from "lock free" — then restore it.

Attribute before claiming a leak is yours (rule 4): snapshot dbsize and the TTL-less key list before and after your change, and compare against the pristine tree's code via git archive HEAD core/ | tar -x -C <scratch>. Never git stash — the stack is shared across worktrees.

Worked example

Real session against the dev container, 2026-07-28:

docker exec devskyy-redis redis-cli ping
docker exec devskyy-redis sh -c 'REDISCLI_AUTH="$REDIS_PASSWORD" redis-cli ping'
docker exec devskyy-redis sh -c '
  REDISCLI_AUTH="$REDIS_PASSWORD" redis-cli set lock:skill-demo:wp-ops tok-1 PX 5000 NX
  REDISCLI_AUTH="$REDIS_PASSWORD" redis-cli set lock:skill-demo:wp-ops tok-2 PX 5000 NX
  REDISCLI_AUTH="$REDIS_PASSWORD" redis-cli pttl lock:skill-demo:wp-ops
  REDISCLI_AUTH="$REDIS_PASSWORD" redis-cli del lock:skill-demo:wp-ops'

Observed [repro]:

NOAUTH Authentication required.
PONG
OK

4985
1

Three facts fall out. The first line is auth working — an agent that reads NOAUTH as "Redis is down" will go fix a container that is healthy. The blank line after OK is the second SET NX returning nil: the lock was already held, so acquisition was correctly refused — that empty line is the entire safety property, and a harness that only checks "did the command run" would miss it. pttl 4985 confirms the 5000ms expiry is armed, which is what stops a crashed holder from wedging the resource forever.

Paired with maxmemory-policy noeviction and maxmemory 0B on this instance, the operative rule for any new key here is the TTL: nothing evicts, nothing is capped, so core/redis_cache.py:191's await self._client.setex(key, ttl, json.dumps(response)) — TTL passed in the same call as the value, defaulting from LLM_CACHE_TTL at core/redis_cache.py:60 [repo] — is the pattern to copy, not a bare set followed by a hopeful expire.

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)

async def get_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 hour
    return 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

@asynccontextmanager
async def lifespan(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)
    yield
    await pool.aclose()

app = FastAPI(lifespan=lifespan)

Write-Through Cache

async def update_product(product_id: int, data: dict) -> None:
    # Write to DB first
    await 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 set
async def cache_product(product_id: int, category_id: int, data: dict) -> None:
    key = f"product:{product_id}"
    tag = f"tag:category:{category_id}"
    async with 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()

async def invalidate_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

async def create_session(user_id: int, ttl: int = 86400) -> str:
    session_id = str(uuid.uuid4())
    key = f"session:{session_id}"
    async with 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

async def get_session(session_id: str) -> dict | None:
    data = await r.hgetall(f"session:{session_id}")
    return data if data else None

async def delete_session(session_id: str) -> None:
    await r.delete(f"session:{session_id}")

Rate Limiting

Fixed Window (Simple)

async def is_rate_limited(user_id: int, limit: int = 100, window: int = 60) -> bool:
    key = f"ratelimit:{user_id}:{int(time.time()) // window}"
    async with r.pipeline(transaction=True) as pipe:
        await pipe.incr(key)
        await pipe.expire(key, window)
        count, _ = await pipe.execute()
    return count > limit

Sliding Window (Lua — Atomic)

-- sliding_window.lua
local key = KEYS[1]
local now = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local limit = tonumber(ARGV[3])

redis.call('ZREMRANGEBYSCORE', key, 0, now - window)
local count = redis.call('ZCARD', key)

if count < limit then
    -- Use unique member (now + sequence) to avoid collisions within the same millisecond
    local seq_key = key .. ':seq'
    local seq = redis.call('INCR', seq_key)
    redis.call('EXPIRE', seq_key, math.ceil(window / 1000))
    redis.call('ZADD', key, now, now .. '-' .. seq)
    redis.call('EXPIRE', key, math.ceil(window / 1000))
    return 1
end
return 0

sliding_window = r.register_script(open('sliding_window.lua').read())

async def allow_request(user_id: int) -> bool:
    key = f"ratelimit:sliding:{user_id}"
    now = int(time.time() * 1000)
    result = await sliding_window(keys=[key], args=[now, 60000, 100])
    return bool(result)

Distributed Locks

Distributed Lock (Single Node — SET NX PX)

import uuid

async def acquire_lock(resource: str, ttl_ms: int = 5000) -> str | None:
    lock_key = f"lock:{resource}"
    token = str(uuid.uuid4())
    acquired = await r.set(lock_key, token, px=ttl_ms, nx=True)
    return token if acquired else None

async def release_lock(resource: str, token: str) -> bool:
    release_script = """
    if redis.call('get', KEYS[1]) == ARGV[1] then
        return redis.call('del', KEYS[1])
    else
        return 0
    end
    """
    result = await r.eval(release_script, 1, f"lock:{resource}", token)
    return bool(result)

# Usage
token = await acquire_lock("order:payment:123")
if token:
    try:
        await process_payment()
    finally:
        await release_lock("order:payment:123", token)

For multi-node setups use the redlock-py library which implements the full Redlock algorithm.

Pub/Sub & Streams

Pub/Sub (Fire-and-Forget)

import asyncio
import redis.asyncio as redis

# Publisher
async def publish_event(channel: str, payload: dict) -> None:
    await r.publish(channel, json.dumps(payload))

# Subscriber — run as a background task (e.g., via asyncio.create_task)
async def subscribe_events(channel: str) -> None:
    async with r.pubsub() as pubsub:
        await pubsub.subscribe(channel)
        while True:
            message = await pubsub.get_message(
                ignore_subscribe_messages=True, timeout=1.0
            )
            if message is not None:
                await handle(json.loads(message["data"]))

Redis Streams (Durable Queue)

# Producer
async def emit(stream: str, event: dict) -> None:
    await r.xadd(stream, event, maxlen=10000)  # Cap stream length

# One-time consumer group setup (idempotent)
async def ensure_consumer_group(stream: str, group: str) -> None:
    try:
        await r.xgroup_create(stream, group, id="0", mkstream=True)
    except Exception:
        pass  # Group already exists

# Consumer — runs in an asyncio task or worker
async def consume(stream: str, group: str, consumer: str) -> None:
    while True:
        messages = await r.xreadgroup(
            group, consumer, {stream: ">"}, count=10, block=2000
        )
        for _, entries in (messages or []):
            for msg_id, data in entries:
                await process(data)
                await r.xack(stream, group, msg_id)

Prefer Streams over Pub/Sub when you need delivery guarantees, consumer groups, or replay.

Key Design

Naming Conventions

# Pattern: resource:id:field
user:123:profile
order:456:status
cache:product:789

# Pattern: namespace:resource:id
myapp:session:abc123
myapp:ratelimit:user:123

# Pattern: resource:date (time-bound keys)
stats:pageviews:2024-01-01

TTL Strategy

Data Type	Suggested TTL
User session	24h (86400)
API response cache	5–15 min
Rate limit window	Match window size
Short-lived tokens	5–10 min
Leaderboard	1h–24h
Static/reference data	1h–1 week

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,
)

Sentinel (High Availability)

from redis.asyncio.sentinel import Sentinel

sentinel = Sentinel(
    [("sentinel-1", 26379), ("sentinel-2", 26379)],
    socket_timeout=0.5,
)
master = sentinel.master_for("mymaster", decode_responses=True)
replica = sentinel.slave_for("mymaster", decode_responses=True)

Eviction Policies

Policy	Behavior	Best For
noeviction	Error on write when full	Queues / critical data
allkeys-lru	Evict least recently used	General cache
volatile-lru	LRU only among keys with TTL	Mixed data store
allkeys-lfu	Evict least frequently used	Skewed access patterns
volatile-ttl	Evict soonest-to-expire	Prioritize long-lived data

Set via redis.conf: maxmemory-policy allkeys-lru

Anti-Patterns

Anti-Pattern	Problem	Fix
Keys with no TTL	Memory grows unbounded	Always set TTL
KEYS * in production	Blocks the server (O(N))	Use SCAN cursor
Storing large blobs (>100KB)	Slow serialization, memory pressure	Store reference + fetch from object store
Single Redis for everything	No isolation between cache & queue	Use separate DBs or instances
Ignoring connection pool limits	Connection exhaustion under load	Size pool to workload
Not handling cache miss stampede	Thundering herd on cold start	Use locks or probabilistic early expiry
FLUSHALL without thought	Wipes entire instance	Scope deletes by key pattern
Using sync import redis in FastAPI handlers	Blocks the event loop	Use redis.asyncio everywhere in async code

Cache Miss Stampede Prevention

import asyncio

_stampede_locks: dict[str, asyncio.Lock] = {}
_locks_mutex = asyncio.Lock()

async def get_with_lock(key: str, fetch_fn, ttl: int = 300):
    cached = await r.get(key)
    if cached:
        return json.loads(cached)

    async with _locks_mutex:
        if key not in _stampede_locks:
            _stampede_locks[key] = asyncio.Lock()
        lock = _stampede_locks[key]

    async with lock:
        cached = await r.get(key)  # Re-check after acquiring lock
        if cached:
            return json.loads(cached)
        value = await fetch_fn()
        await r.setex(key, ttl, json.dumps(value))
        return value

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
• Skill: database-migrations — schema versioning
• Skill: django-patterns — Django cache framework integration
• Agent: database-reviewer — full database review workflow