Redis: Strings, Hashes, Sorted Sets and Streams Reference Guide

# macOS — Homebrew
brew install redis
brew services start redis

# Linux — Debian/Ubuntu
sudo apt install redis-server
sudo systemctl enable --now redis-server

# Docker — disposable dev instance
docker run --name redis -p 6379:6379 -d redis:7

# Connect (CLI)
redis-cli                                   # localhost:6379
redis-cli -h prod.example -p 6380 -a $PW    # remote + auth
redis-cli -u "rediss://default:$PW@host:6380/0"   # URL form (TLS)

# Python — modern client (sync + async in one package)
pip install "redis>=5.0"

# Test the connection
redis-cli PING
# → PONG

# Append an event ("*" → server-generated id)
XADD orders * sku abc qty 3
# → "1715944800123-0"

# Create a consumer group starting from the current end ("$")
XGROUP CREATE orders billing '$' MKSTREAM

# A worker reads new events (blocks up to 5s)
XREADGROUP GROUP billing worker-1 COUNT 10 BLOCK 5000 STREAMS orders '>'

# Acknowledge after processing
XACK orders billing 1715944800123-0

# See still-pending (claimed but un-ACKed) entries
XPENDING orders billing
XAUTOCLAIM orders billing worker-2 60000 0    # steal stuck messages

# Cap stream length (lossy trim — keeps newest ~10k)
XADD orders MAXLEN '~' 10000 * sku abc qty 3

# Range scan (useful for replay)
XRANGE orders - + COUNT 100

-- Safe lock release: only delete if the value still matches our token.
-- Run via EVAL — KEYS[1] = lock key, ARGV[1] = token.
if redis.call("GET", KEYS[1]) == ARGV[1] then
  return redis.call("DEL", KEYS[1])
else
  return 0
end

from redis import Redis
from redis.asyncio import Redis as AsyncRedis

# Sync — connection pool is created for you
r = Redis(host="localhost", port=6379, db=0, decode_responses=True)

r.set("user:42:name", "Ana", ex=3600)        # TTL 1h
r.get("user:42:name")                         # "Ana"

# Pipeline — batch commands, one round-trip
with r.pipeline(transaction=False) as pipe:
    pipe.hset("user:42", mapping={"name": "Ana", "plan": "pro"})
    pipe.expire("user:42", 86400)
    pipe.zadd("leaderboard", {"42": 1500})
    pipe.execute()

# Transaction — WATCH for optimistic concurrency
def increment_if_below(key, cap):
    with r.pipeline() as pipe:
        while True:
            try:
                pipe.watch(key)
                cur = int(pipe.get(key) or 0)
                if cur >= cap:
                    pipe.unwatch(); return cur
                pipe.multi()
                pipe.incr(key)
                pipe.execute()
                return cur + 1
            except r.WatchError:
                continue   # somebody else wrote — retry

# Async
async def main():
    a = AsyncRedis.from_url("redis://localhost", decode_responses=True)
    await a.set("k", "v"); print(await a.get("k"))
    await a.aclose()

# Tiny rate-limiter: N requests per minute per user.
# Uses INCR + EXPIRE — both run atomically in one pipeline.
from redis import Redis

r = Redis(decode_responses=True)

def allow(user_id: str, limit: int = 60, window_sec: int = 60) -> bool:
    key = f"rl:{user_id}:{int(time.time()) // window_sec}"
    with r.pipeline() as pipe:
        pipe.incr(key)
        pipe.expire(key, window_sec)
        count, _ = pipe.execute()
    return count <= limit

# Leaderboard with sorted set
r.zadd("leaderboard", {"alice": 1500, "bob": 1820, "carol": 1730})
r.zrevrange("leaderboard", 0, 2, withscores=True)
# → [('bob', 1820.0), ('carol', 1730.0), ('alice', 1500.0)]

# Cache-aside read with TTL
def get_user(user_id):
    key = f"user:{user_id}"
    cached = r.get(key)
    if cached: return json.loads(cached)
    row = db.fetch_user(user_id)             # cache miss → source of truth
    r.set(key, json.dumps(row), ex=300)      # cache 5 min
    return row