Redis Patterns: Cache, Sessions, Queues, Locks

Redis is an in-memory store for many kinds of data structure. Its own description is "data structure server", and that is exactly what it gives you: strings, hashes, lists, sets, sorted sets and streams. Every command is atomic, runs in microseconds over a TCP connection, and works on data held in RAM. Most people first use Redis for caching. Soon after, they use it for sessions, queues, rate limiters, distributed locks, leaderboards and pub/sub too.

This lesson covers the patterns used in production: when to use each data type, how TTLs stop Redis from filling your RAM, and the common mistakes that look fine in a demo but cause problems in production.

Examples assume a running local Redis. Docker is the easiest way to get one. 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 lookup, even with everything cached, is several times slower, because it has to parse SQL, apply MVCC rules and walk B-trees. For data that is read constantly and changes rarely (session data, rendered pages, computed totals), Redis is a good fit.

Redis is more than a key-value store. The same instance can serve as a:

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, so a large part of using Redis well is choosing the right one.

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 you no longer need to call .decode() everywhere. For binary payloads (protobuf, msgpack, images) leave it off and handle bytes directly.

In production you use a connection pool. redis-py creates one automatically, but you can configure it:

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

The async client is in redis.asyncio. It has the same API, but you await each call:

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

Use it in async code (FastAPI, async workers), and never call the sync redis client from a coroutine. This is the same rule as for any blocking call. See async.

3. Strings: The Most Used Type

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, and 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: Changing One Field at a Time

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 are most useful 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 that Redis stores every value as a string, so convert it when you read it (int(r.hget("user:1", "age"))).

5. Lists: Queues and Activity Logs

Redis lists are linked lists. LPUSH to the head and RPOP from the tail (or the reverse) gives you a FIFO queue. BLPOP is the blocking version: it waits until a value appears. Most simple home-made job queues are built on it.

python

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

# Worker side: wait 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 ten most recent items, which is useful for activity feeds.

For real queues with retries, scheduling, and visibility timeouts, use RQ, Celery or Streams (Section 7) instead of writing your own on plain lists.

6. Sets and Sorted Sets

Sets are unordered collections with no duplicates. Membership tests are 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: users who are online AND premium

+ 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 by score. They are the data structure behind leaderboards, sliding-window rate limits and time-ordered 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 (ranks count from 0, lowest score first)
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 simplest way to build a sliding-window rate limiter (Section 10).

7. Streams: An Event Log

Streams (XADD, XREAD, XREADGROUP) are an append-only log with consumer groups. For smaller workloads, they do a similar job to Kafka.

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

This is the pattern you will use most often. Check Redis. On a miss, fetch the value from the source of truth, write it to Redis with a TTL, and return it.

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 old can the data be? 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? Deleting on each change keeps the cache fresher, but it is harder to get right.
3. Key naming: use colon-separated namespaces: user:1:posts, cache:product:42, session:<token>. Consistent names let you delete a whole group of keys with SCAN + DEL.

A harder problem is the cache stampede: a popular key expires, and a thousand requests all miss at the same moment and all hit 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 it 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: check and delete in one atomic step, so we never free another client'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 while holding the lock, the lock expires automatically.
Token identifies the holder. Before releasing, the code checks "is this lock 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 most use cases.

The redis-py library also ships r.lock(name, timeout=...) which does all of 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)              # so the key does not stay in Redis forever
    _, 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 traffic, switch to fixed-window counters with INCR + EXPIRE. They are less accurate, but much cheaper.

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 lost. 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. A hundred commands sent one after another add 20 ms of waiting that you can avoid.

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

r.pipeline() is transactional by default: redis-py wraps the batch in MULTI/EXEC, so it runs atomically. Pass transaction=False for a plain buffered batch (one round trip, no atomicity).

For check-then-act semantics across keys, use Lua scripts (r.eval(...)). Redis runs each script atomically on the server.

13. Persistence and Eviction

Redis keeps data in memory, but it can also save it to disk.

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	A common choice for production: you get the benefits 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)

Choose one yourself. With the default, noeviction, Redis returns OOM command not allowed once the instance is full.

14. Scaling Beyond One Node

Replication: one primary and N read replicas. Easy and common, and it handles more reads.
Sentinel: adds failover when the primary dies.
Redis Cluster: horizontal sharding across multiple primaries by hash slot. You need it once your data won't fit on one node, but multi-key operations (commands such as MGET or SUNION, transactions and Lua scripts) only work when all their keys are in a single hash slot. Use hash tags ({user:1}:posts) to colocate related keys.

Most apps run well on a single replicated instance for years before they need a cluster.

Common Mistakes

1. No TTL on cache keys. Memory keeps growing until the eviction policy starts dropping keys you didn't choose, or you hit OOM. Set a TTL on every cache write.

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 common, and it makes keys hard to find and delete. 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 return values that can't be JSON-serialised. 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. On a hit, return the deserialised value
            # TODO 3: on a 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, then prints 13
print(expensive_calc(3, 4))               # cache hit, so it only prints 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, but JSON with sorted keys is. Use default=str so that non-JSON types don't raise an error.

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.

    Key: the function name plus a sha1 of (args, kwargs). The result is stored as JSON.
    A result that can't be stored as JSON is returned 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))               # prints "computing for 3,4", then 13
print(expensive_calc(3, 4))               # 13 (cache hit; no compute line)
print(expensive_calc(5, 6))               # prints "computing for 5,6", then 31

What the solution gets right:

Stable key: json.dumps(..., sort_keys=True) gives the same string across runs and processes. repr() or string-concatenation would not be.
One-key envelope {"v": result}: tells apart a cached None from a cache miss. Without it, you'd recompute every time the function legitimately returns None.
Namespace prefix cache:: easy to delete all at once with SCAN cache:* + DEL.
Skips non-serialisable returns: a function returning a datetime or a custom class still works. Its result is just not cached.
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: it stores strings, hashes, lists, sets, sorted sets and streams, not only key-value pairs.
Connection pools are automatic but tunable. decode_responses=True saves you from calling .decode() everywhere.
TTLs on every cache key: set(..., ex=seconds), so memory can't grow forever.
Cache-aside is the most common pattern: check, miss, fetch, write with TTL, return.
Distributed locks via SET NX EX with a unique token, and release via Lua script for atomicity.
Sliding-window rate limits via sorted sets and timestamps in a pipeline.
Pub/Sub is fire-and-forget, but Streams are durable and replayable.
Pipelines send N commands in one round trip, which makes batched writes much faster.
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: data like invoices still belongs in Postgres.

Next: MongoDB: when a document store is the right choice, and when Postgres JSONB does the same job better.

1. Why Redis

2. Connecting

3. Strings: The Most Used Type

4. Hashes: Changing One Field at a Time

5. Lists: Queues and Activity Logs

6. Sets and Sorted Sets

7. Streams: An 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