Redis — Patterns: Caching, Rate Limits, Locks, Messaging¶

import redis

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

# app/cache.py
import json
from collections.abc import Callable

import redis

USER_TTL = 300                                    # seconds


def user_key(user_id: int) -> str:
    return f"myapp:user:{user_id}"


def get_user(r: redis.Redis, user_id: int, load: Callable[[int], dict]) -> dict:
    key = user_key(user_id)
    cached = r.get(key)
    if cached is not None:                        # hit
        return json.loads(cached)
    user = load(user_id)                          # miss: read the source of truth
    r.set(key, json.dumps(user), ex=USER_TTL)
    return user


def update_user(r: redis.Redis, user_id: int, data: dict, save: Callable[[int, dict], None]) -> None:
    save(user_id, data)                           # 1. write the database
    r.delete(user_key(user_id))                   # 2. drop the cached copy

import random


def cache_set(r: redis.Redis, key: str, value: str, ttl: int = 300) -> None:
    r.set(key, value, ex=ttl + random.randint(0, ttl // 10))   # jitter: keys written together do not expire together

# app/ratelimit.py
import time
from collections.abc import Callable

import redis


def allow(
    r: redis.Redis,
    subject: str,
    limit: int,
    window_s: int,
    now: Callable[[], float] = time.time,         # injectable clock for tests
) -> bool:
    window = int(now() // window_s)
    key = f"myapp:rl:{subject}:{window}"
    pipe = r.pipeline()                           # MULTI/EXEC by default
    pipe.incr(key)
    pipe.expire(key, window_s, nx=True)           # set TTL only on the first hit (Redis 7.0+)
    count, _ = pipe.execute()
    return count <= limit

import time
import uuid

SLIDING_WINDOW = r.register_script("""
local key, now_ms, window_ms, limit = KEYS[1], tonumber(ARGV[1]), tonumber(ARGV[2]), tonumber(ARGV[3])
redis.call('ZREMRANGEBYSCORE', key, 0, now_ms - window_ms)
if redis.call('ZCARD', key) >= limit then
  return 0
end
redis.call('ZADD', key, now_ms, ARGV[4])
redis.call('PEXPIRE', key, window_ms)
return 1
""")


def allow_sliding(r: redis.Redis, subject: str, limit: int, window_s: int) -> bool:
    now_ms = int(time.time() * 1000)
    member = uuid.uuid4().hex                     # unique member: two requests in the same ms both count
    return SLIDING_WINDOW(keys=[f"myapp:rl:sliding:{subject}"], args=[now_ms, window_s * 1000, limit, member]) == 1


[allow_sliding(r, "user:1", 3, 10) for _ in range(5)]   # [True, True, True, False, False]

import uuid

RELEASE = r.register_script("""
if redis.call('GET', KEYS[1]) == ARGV[1] then
  return redis.call('DEL', KEYS[1])
end
return 0
""")


def acquire(r: redis.Redis, name: str, ttl_ms: int) -> str | None:
    token = uuid.uuid4().hex
    return token if r.set(f"myapp:lock:{name}", token, nx=True, px=ttl_ms) else None


def release(r: redis.Redis, name: str, token: str) -> bool:
    return RELEASE(keys=[f"myapp:lock:{name}"], args=[token]) == 1   # delete only our own lock

from redis.exceptions import LockError

with r.lock("myapp:lock:import", timeout=30, blocking_timeout=5) as lock:   # LockError if not acquired in 5 s
    run_import()
    lock.extend(30)                               # add 30 s to the remaining TTL for a long step
    finish_import()

lock = r.lock("myapp:lock:nightly-report", timeout=60, blocking=False)
if lock.acquire():                                # False at once if someone else holds it
    try:
        build_report()
    finally:
        lock.release()                            # LockNotOwnedError if the lock already expired

# Pub/Sub: fire-and-forget
sub = r.pubsub()
sub.subscribe("myapp:events:orders")
r.publish("myapp:events:orders", '{"id": 1}')      # returns the number of receivers: 1
r.publish("myapp:events:orders", "lost")           # 0 receivers if nobody is subscribed: the message is dropped

# Stream with a consumer group: at-least-once
r.xgroup_create("myapp:orders", "billing", id="0", mkstream=True)
r.xadd("myapp:orders", {"order_id": 1}, maxlen=100_000, approximate=True)
for stream, entries in r.xreadgroup("billing", "worker-1", {"myapp:orders": ">"}, count=10, block=5000):
    for entry_id, fields in entries:
        handle(fields)
        r.xack("myapp:orders", "billing", entry_id)     # not acked → stays in XPENDING, can be reclaimed

MULTI
INCR myapp:orders:count
EXPIRE myapp:orders:count 3600
EXEC            -- 1) (integer) 1  2) (integer) 1

with r.pipeline() as pipe:                        # transaction=True by default
    pipe.set("s", "text")
    pipe.incr("s")                                # will fail on the server
    pipe.set("after", 1)
    pipe.execute()                                # raises ResponseError — but "s" and "after" were written
# pipe.execute(raise_on_error=False) → [True, ResponseError(...), True]

from redis.exceptions import WatchError


def withdraw(r: redis.Redis, account: int, amount: int) -> bool:
    key = f"myapp:balance:{account}"
    with r.pipeline() as pipe:
        while True:
            try:
                pipe.watch(key)                   # from here commands run immediately
                balance = int(pipe.get(key) or 0)
                if balance < amount:
                    pipe.unwatch()
                    return False
                pipe.multi()                      # from here commands are queued
                pipe.decrby(key, amount)
                pipe.execute()                    # WatchError if key changed since WATCH
                return True
            except WatchError:
                continue                          # another client wrote first: retry

with r.pipeline(transaction=False) as pipe:       # no MULTI/EXEC, just batching
    for i in range(1000):
        pipe.set(f"myapp:item:{i}", i, ex=3600)
    results = pipe.execute()                      # list of 1000 replies

# EVAL: keys first, then arguments
r.eval("return redis.call('INCRBY', KEYS[1], ARGV[1])", 1, "myapp:counter", 5)   # 5

# register_script: sends EVALSHA, falls back to EVAL on NOSCRIPT (after a restart or SCRIPT FLUSH)
cas = r.register_script("""
if redis.call('GET', KEYS[1]) == ARGV[1] then
  redis.call('SET', KEYS[1], ARGV[2])
  return 1
end
return 0
""")
cas(keys=["myapp:config:mode"], args=["v1", "v2"])