Redis Patterns: Cache, Sessions, Queues, Locks

Redis is the Swiss army knife of in-memory data structures. The official tagline is "data structure server", and that's literally what you're paying for — strings, hashes, lists, sets, sorted sets, streams — all atomic, all accessed over a single TCP connection in microseconds, all in RAM. You will use it for caching first, because that's everyone's gateway, then for sessions, queues, rate limiters, distributed locks, leaderboards, and pub/sub before you know it.

This lesson is the production patterns — when to reach for each data type, the TTL discipline that keeps Redis from eating your RAM, and the six anti-patterns that look fine in a demo and bite in production.

Examples assume a running local Redis — Docker is the easiest way; see devops-docker. Pip-install the driver as shown. Expected output is in comments.

1. Why Redis

Redis lives in RAM. A GET key takes ~100 microseconds locally, ~1 millisecond across a data centre. Postgres on the same query, even with everything cached, is 5-10× slower because it has to honour MVCC, parse SQL, and walk B-trees. For data that's read constantly and changes rarely — session blobs, rendered pages, computed aggregates — Redis is the right answer.

It's also a tool kit, not just a key-value store. The same instance can serve:

Cache layer in front of a slow database or API
Session store for stateless web servers
Rate limiter with sliding-window counters
Distributed lock for cross-process mutual exclusion
Task queue with BLPOP or Streams
Leaderboard with sorted sets
Pub/Sub for fan-out notifications

Each pattern uses a different data type. Knowing which is half the skill.

2. Connecting

bash

pip install redis

python

import redis

r = redis.Redis(host="localhost", port=6379, decode_responses=True)
r.set("greeting", "hello")
print(r.get("greeting"))                  # hello

decode_responses=True tells the driver to return str instead of bytes. Set it once and stop calling .decode() everywhere. For binary payloads (protobuf, msgpack, images) leave it off and handle bytes directly.

For production, use a connection pool — it's automatic but configurable:

python

pool = redis.ConnectionPool(host="localhost", port=6379, max_connections=50, decode_responses=True)
r = redis.Redis.from_pool(pool)

+ setup added so this can run · defines redis

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

redis = _AutoMock('redis')

Async lives at redis.asyncio — same API, awaitable:

python

import redis.asyncio as aioredis

async def main():
    r = aioredis.Redis(host="localhost", port=6379, decode_responses=True)
    await r.set("greeting", "hello")
    print(await r.get("greeting"))        # hello
    await r.aclose()

Drop in async-native code (FastAPI, async workers) and stop using sync redis from a coroutine — same advice as for any blocking call. See async.

3. Strings — The Workhorse

Strings hold up to 512 MB each. Beyond plain bytes, Redis treats them as counters when appropriate.

python

r.set("counter", 0)
r.incr("counter")                         # 1
r.incrby("counter", 10)                   # 11
r.decr("counter")                         # 10

r.set("session:abc", "{...json...}", ex=3600)    # TTL of 1 hour
r.ttl("session:abc")                      # 3600 — seconds left

+ setup added so this can run · defines r

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

r = _AutoMock('r')

ex= sets seconds-to-live; px= does milliseconds; exat= and pxat= set absolute Unix timestamps. Always set a TTL on cache and session keys — see Common Mistakes.

SET key value NX EX 30 is the atomic "set if not exists with TTL" primitive that builds a distributed lock; we use it in Section 9.

4. Hashes — Field-Level Mutations

When your value is a small struct with named fields, a hash beats serialising to JSON and re-parsing on every update.

python

r.hset("user:1", mapping={"name": "Alice", "age": "30", "active": "1"})
r.hget("user:1", "name")                  # 'Alice'
r.hgetall("user:1")                       # {'name': 'Alice', 'age': '30', 'active': '1'}
r.hincrby("user:1", "login_count", 1)     # atomic increment of one field

+ setup added so this can run · defines r

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

r = _AutoMock('r')

Hashes shine when you update one field often — HINCRBY and HSET change one field in O(1), without touching the rest. They're also memory-efficient for many small structs.

Note Redis values are stringly-typed; cast on the way out (int(r.hget("user:1", "age"))).

5. Lists — Queues and Activity Logs

Lists are doubly-linked lists. LPUSH to the head, RPOP from the tail (or the reverse) gives you a FIFO queue. BLPOP is the blocking variant that waits for a value to appear — the basis of every "poor man's job queue".

python

r.lpush("jobs", '{"task": "send_email", "to": "alice@example.com"}')
r.lpush("jobs", '{"task": "render_pdf", "id": 42}')

# Worker side — block up to 5 seconds for the next job
job = r.brpop("jobs", timeout=5)
# ('jobs', '{"task": "send_email", ...}')

+ setup added so this can run · defines r

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

r = _AutoMock('r')

LRANGE list 0 9 returns the most recent ten items — useful for activity feeds.

For real queues with retries, scheduling, and visibility timeouts, reach for RQ, Celery, or Streams (Section 7) instead of rolling your own on bare lists.

6. Sets and Sorted Sets

Sets are unordered, deduplicated collections — membership tests in O(1):

python

r.sadd("users:online", "1", "2", "3")
r.sismember("users:online", "2")          # True
r.scard("users:online")                   # 3
r.sinter("users:online", "users:premium") # set intersection — for "online premium users"

+ setup added so this can run · defines r

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

r = _AutoMock('r')

Sorted sets map members to floating-point scores and keep them ordered — the data structure behind leaderboards, sliding-window rate limits, and time-bucketed queues:

python

r.zadd("leaderboard", {"surya": 1200, "alex": 950, "kira": 1480})
r.zrevrange("leaderboard", 0, 2, withscores=True)
# [('kira', 1480.0), ('surya', 1200.0), ('alex', 950.0)]

r.zrank("leaderboard", "surya")           # 1 (0-indexed from low)
r.zincrby("leaderboard", 50, "surya")     # 1250

+ setup added so this can run · defines r

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

r = _AutoMock('r')

Sorted sets are also the cleanest sliding-window rate limiter (Section 10).

7. Streams — The Modern Event Log

Streams (XADD, XREAD, XREADGROUP) are an append-only log with consumer-group semantics — Redis's answer to Kafka for smaller workloads.

python

r.xadd("events", {"type": "click", "user": "1"})
r.xadd("events", {"type": "purchase", "user": "1", "amount": "42"})

entries = r.xread({"events": "0"}, count=10)
# [['events', [('1700000000-0', {'type': 'click', 'user': '1'}), ...]]]

+ setup added so this can run · defines r

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

r = _AutoMock('r')

Use streams when you want durability, at-least-once delivery, multiple consumers with progress tracking, and ordered events. Use plain lists when you want lightweight fire-and-forget queues.

8. The Cache-Aside Pattern

The 80% pattern. Check Redis; on a miss, fetch from the source of truth, write to Redis with a TTL, return.

python

import json

def get_user(user_id: int) -> dict:
    key = f"user:{user_id}"
    cached = r.get(key)
    if cached is not None:
        return json.loads(cached)

    user = fetch_from_postgres(user_id)        # the expensive call
    r.set(key, json.dumps(user), ex=300)       # 5-minute TTL
    return user

+ setup added so this can run · defines fetch_from_postgres, r

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

def fetch_from_postgres(*_a, **_kw):
    print('-> fetch_from_postgres() called')
    return _AutoMock('fetch_from_postgres()')
r = _AutoMock('r')

Three knobs to think about per cache key:

1. TTL — how stale is acceptable? Sessions: 30 minutes. Product catalogues: 1 hour. Static reference data: 24 hours.
2. Invalidation — when the underlying data changes, do you DEL the key, or just wait for TTL? Event-based invalidation is faster but harder to keep right.
3. Key naming — colon-separated namespaces: user:1:posts, cache:product:42, session:<token>. Consistency makes it possible to bulk-delete via SCAN + DEL.

The harder problem is cache stampede — a hot key expires and a thousand requests all miss simultaneously, hammering the database. Mitigations: probabilistic early expiration, singleflight (only one in-flight fetch per key), or pre-warming on background workers.

9. Distributed Locks

SET key value NX EX seconds is atomic — it succeeds only if the key didn't exist, and sets a TTL. That's enough for a basic distributed mutex.

python

import uuid, time

def acquire_lock(name: str, ttl: int = 30) -> str | None:
    token = uuid.uuid4().hex
    if r.set(f"lock:{name}", token, nx=True, ex=ttl):
        return token
    return None

def release_lock(name: str, token: str) -> bool:
    # Lua script — atomic check-and-delete so we don't release someone else's lock
    script = """
    if redis.call("get", KEYS[1]) == ARGV[1] then
        return redis.call("del", KEYS[1])
    else
        return 0
    end
    """
    return bool(r.eval(script, 1, f"lock:{name}", token))

token = acquire_lock("import-job", ttl=60)
if token:
    try:
        run_the_job()
    finally:
        release_lock("import-job", token)

+ setup added so this can run · defines r, run_the_job

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

r = _AutoMock('r')
def run_the_job(*_a, **_kw):
    print('-> run_the_job() called')
    return _AutoMock('run_the_job()')

Two things matter:

TTL is the safety net — if the process dies holding the lock, it expires automatically.
Token identifies the holder — releasing checks "is this still mine?" so a slow process whose lock already expired can't release someone else's.

For correctness under network partitions and replica failover, see Redlock. The single-instance version above is fine for 99% of use cases.

The redis-py library also ships r.lock(name, timeout=...) which wraps this for you.

10. Sliding-Window Rate Limiter

Sorted sets with timestamps make rate limiting straightforward and accurate:

python

import time

def allow(user_id: int, *, limit: int = 100, window: float = 60.0) -> bool:
    key = f"ratelimit:{user_id}"
    now = time.time()
    pipe = r.pipeline()
    pipe.zremrangebyscore(key, 0, now - window)   # drop entries older than the window
    pipe.zcard(key)                                # count what's left
    pipe.zadd(key, {f"{now}-{uuid.uuid4()}": now}) # record this request
    pipe.expire(key, int(window) + 1)              # keep the key from leaking
    _, count, _, _ = pipe.execute()
    return count < limit

+ setup added so this can run · defines r, uuid

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

r = _AutoMock('r')
uuid = _AutoMock('uuid')

One sliding window per user, fully atomic via a pipeline, accurate to milliseconds. For very high-throughput cases, switch to fixed-window counters with INCR + EXPIRE (less accurate, ~10× faster).

11. Pub/Sub

python

# Subscriber (in another process)
sub = r.pubsub()
sub.subscribe("notifications")
for msg in sub.listen():
    if msg["type"] == "message":
        print("got:", msg["data"])

# Publisher
r.publish("notifications", "hello, surya")

+ setup added so this can run · defines r

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

r = _AutoMock('r')

Pub/Sub is fire-and-forget — if no subscriber is listening, the message is gone. For durable, replayable streams, use Redis Streams (Section 7) or a real broker (Kafka, RabbitMQ).

12. Pipelines — One Round Trip, Many Commands

Each Redis command costs one network round trip — perhaps 200 microseconds in a data centre. Issuing a hundred of them serially is 20 ms of latency you don't have to pay.

python

with r.pipeline() as pipe:
    for i in range(100):
        pipe.set(f"counter:{i}", 0)
    pipe.execute()
# One TCP round trip; ~100× faster than 100 separate set() calls.

+ setup added so this can run · defines r

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

r = _AutoMock('r')

Pipelines are not transactions — by default they're just buffered batches. Pass transaction=True (or use pipe.multi() / pipe.execute()) for an atomic MULTI/EXEC block.

For check-then-act semantics across keys, use Lua scripts (r.eval(...)) — Redis executes them atomically server-side.

13. Persistence and Eviction

Redis is in-memory but durable when you want it.

Mode	Trade-off
RDB (snapshots)	Periodic dumps; lose at most N minutes on crash; fast restarts.
AOF (append-only file)	Every write logged; near-zero data loss; slower restarts.
RDB + AOF	The default for production — best of both.
No persistence	Pure cache; fine if you can rebuild from the source.

When memory hits maxmemory, an eviction policy decides what to drop:

allkeys-lru — least-recently-used across all keys (good for pure cache instances)
volatile-lru — only evict keys with a TTL (when you mix cache and "permanent" data)
allkeys-lfu — least-frequently-used
noeviction — refuse new writes (data store mode)

Pick one explicitly. The default noeviction will cheerfully return OOM command not allowed when you fill the instance.

14. Scaling Beyond One Node

Replication — one primary, N read replicas. Easy, common, handles read scale.
Sentinel — adds failover when the primary dies.
Redis Cluster — horizontal sharding across multiple primaries by hash slot. Required once your working set won't fit on one node, but multi-key operations (transactions, pipelines, sets across keys) only work within a single hash slot. Use hash tags ({user:1}:posts) to colocate related keys.

Most apps live happily on a single replicated instance for years before they need cluster.

Common Mistakes

1. No TTL on cache keys. Memory grows monotonically until the eviction policy starts dropping random things, or you hit OOM. Set a TTL on every cache write. Always.

2. Using Redis as the system of record. Redis is fast and durable enough for sessions and counters, but it is not your billing database. Anything you can't afford to lose lives in Postgres (or equivalent). Redis is the cache and the queue, not the source of truth.

3. Sync Redis inside async code. Blocks the event loop. Use redis.asyncio.Redis in async services. Same rule as async.

4. No key naming convention. user_1_posts, users:1:posts, posts.user.1 — three formats in one codebase is normal and ruinous. Pick namespace:id:subkey and enforce it.

5. KEYS * in production. KEYS walks every key and blocks the server while it does. On a million-key instance, that's a multi-second freeze for every client. Use SCAN (cursor-based, non-blocking) for iteration.

6. Caching mutable references. Caching a database row that you also update via writes-to-Postgres without invalidating the cache → stale reads forever. Invalidate on write, or accept TTL-bounded staleness, or use a write-through cache.

7. Storing huge values. A 50 MB JSON blob per key, multiplied across keys, fills RAM fast and slows down every operation that touches it. Compress, or store the blob in S3 and cache only the metadata.

🎯 Your Turn — `@cache_with_redis` Decorator

Write a decorator cache_with_redis(ttl=60) that caches the return value of any function in Redis. Cache key = function name + a stable hash of its arguments. JSON-serialise the result.

Requirements:

Apply to any function with hashable args (positional and keyword).
Cache hit → return cached value, no function call.
Cache miss → call the function, store the result with the given TTL, return it.
Use a namespace prefix cache: to keep things tidy.
Don't crash on non-JSON-serialisable return values — log a warning and skip caching.

Skeleton:

python

import functools
import hashlib
import json
import logging

import redis

r = redis.Redis(host="localhost", port=6379, decode_responses=True)

def cache_with_redis(ttl: int = 60):
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            # TODO 1: build a deterministic key from fn.__name__, args, kwargs
            # TODO 2: GET the key — if hit, return the deserialised value
            # TODO 3: MISS — call fn, store the result with ex=ttl, return it
            ...
        return wrapper
    return decorator

@cache_with_redis(ttl=30)
def expensive_calc(x, y):
    print(f"  computing for {x},{y}")
    return x * y + 1

print(expensive_calc(3, 4))               # computes; prints 13
print(expensive_calc(3, 4))               # cache hit; just 13

Hint 1 — Stable argument hashing

JSON-dump the (args, sorted kwargs) tuple with sort_keys=True and hash that string with hashlib.sha1. repr() is not stable for dicts on every Python version — JSON is. Use default=str to handle non-JSON types gracefully.

Hint 2 — Sentinel for the miss path

r.get(key) returns None on a miss and when the cached value is JSON null. Distinguish with r.exists(key) first, or store every value wrapped in a one-key object like {"v": ...}.

Show full solution

python

import functools
import hashlib
import json
import logging

import redis

logger = logging.getLogger(__name__)

r = redis.Redis(host="localhost", port=6379, decode_responses=True)


def _make_key(fn_name: str, args: tuple, kwargs: dict) -> str:
    payload = json.dumps([args, sorted(kwargs.items())], sort_keys=True, default=str)
    digest = hashlib.sha1(payload.encode()).hexdigest()[:16]
    return f"cache:{fn_name}:{digest}"


def cache_with_redis(ttl: int = 60):
    """Cache the return value of `fn` in Redis for `ttl` seconds.

    Cache key = function name + sha1 of (args, kwargs). Result is JSON-encoded.
    Non-serialisable returns are passed through uncached with a warning.
    """
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            key = _make_key(fn.__name__, args, kwargs)

            cached = r.get(key)
            if cached is not None:
                return json.loads(cached)["v"]

            result = fn(*args, **kwargs)

            try:
                r.set(key, json.dumps({"v": result}), ex=ttl)
            except (TypeError, ValueError) as exc:
                logger.warning("cache skip for %s: %s", fn.__name__, exc)

            return result

        return wrapper
    return decorator


# Demo
@cache_with_redis(ttl=30)
def expensive_calc(x, y):
    print(f"  computing for {x},{y}")
    return x * y + 1


print(expensive_calc(3, 4))               # computing for 3,4 → 13
print(expensive_calc(3, 4))               # 13 (cache hit; no compute line)
print(expensive_calc(5, 6))               # computing for 5,6 → 31

What the solution gets right:

Stable key — json.dumps(..., sort_keys=True) is deterministic across runs and processes. repr() or string-concatenation would not be.
One-key envelope {"v": result} — distinguishes a cached None from a cache miss. Without it, you'd recompute every time the function legitimately returns None.
Namespace prefix cache: — easy to bulk-purge with SCAN cache:* + DEL.
Graceful skip on non-serialisable returns — a function returning a datetime or a custom class doesn't crash; it just doesn't cache.
TTL — every key has one, so memory cannot grow forever.

Production extensions you'd add: a circuit breaker that bypasses Redis on driver errors, async support (redis.asyncio with @functools.wraps on an async def wrapper), and per-call TTL overrides via a ttl kwarg sentinel.

What You Learned

Redis is a data-structure server — strings, hashes, lists, sets, sorted sets, streams — not just a key-value store.
Connection pools are automatic but tunable; decode_responses=True saves a lot of .decode() clutter.
TTLs on every cache key — set(..., ex=seconds). Memory growth is your enemy.
Cache-aside is the 80% pattern: check, miss, fetch, write with TTL, return.
Distributed locks via SET NX EX with a unique token; release via Lua script for atomicity.
Sliding-window rate limits via sorted sets and timestamps in a pipeline.
Pub/Sub is fire-and-forget; Streams are durable and replayable.
Pipelines collapse N commands into one round trip — huge speed-up for batched writes.
Persistence: RDB + AOF for production; pick an eviction policy that matches your use (allkeys-lru for pure cache).
Async with redis.asyncio — never call sync Redis from a coroutine.
Not the system of record — Postgres still owns your invoices.

Next: MongoDB — when a document store is genuinely the right call, and when Postgres JSONB has eaten its lunch.

1. Why Redis

2. Connecting

3. Strings — The Workhorse

4. Hashes — Field-Level Mutations

5. Lists — Queues and Activity Logs

6. Sets and Sorted Sets

7. Streams — The Modern Event Log

8. The Cache-Aside Pattern

9. Distributed Locks

10. Sliding-Window Rate Limiter

11. Pub/Sub

12. Pipelines — One Round Trip, Many Commands

13. Persistence and Eviction

14. Scaling Beyond One Node

Common Mistakes

🎯 Your Turn — @cache_with_redis Decorator

What You Learned

🎯 Your Turn — `@cache_with_redis` Decorator