Redis Cheatsheet 19 — Common Use Cases

def get_user(id):
    key = f"user:{id}"
    cached = redis.get(key)
    if cached: return json.loads(cached)
    user = db.user.find(id)
    redis.setex(key, 300, json.dumps(user))
    return user

sid = secrets.token_urlsafe(32)
redis.setex(f"session:{sid}", 86400 * 7, json.dumps({"user_id": uid}))

def allow(key, limit, window):
    pipe = redis.pipeline()
    pipe.incr(key); pipe.expire(key, window)
    count, _ = pipe.execute()
    return count <= limit

redis.xadd("jobs", {"task": "process", "id": 42})

# Worker
res = redis.xreadgroup("workers", "w1", {"jobs": ">"}, count=1, block=5000)

redis.zadd("scores", {"alice": 100, "bob": 85})
top10 = redis.zrevrange("scores", 0, 9, withscores=True)
my_rank = redis.zrevrank("scores", "alice")

# Add post to user's feed
def post(user_id, post_id, ts):
    for follower in get_followers(user_id):
        redis.lpush(f"feed:{follower}", post_id)
        redis.ltrim(f"feed:{follower}", 0, 999)        # keep top 1000

def get_feed(user_id, limit=20):
    return redis.lrange(f"feed:{user_id}", 0, limit - 1)

redis.lpush(f"notifs:{user_id}", json.dumps(notif))
redis.ltrim(f"notifs:{user_id}", 0, 99)
redis.publish(f"user:{user_id}", json.dumps(notif))  # if user online

def heartbeat(user_id):
    redis.zadd("online", {user_id: time.time()})

def cleanup():
    cutoff = time.time() - 60
    redis.zremrangebyscore("online", 0, cutoff)

def online_count():
    return redis.zcard("online")

def increment_view(post_id):
    redis.incr(f"views:{post_id}")

def daily_views(post_id, date):
    return redis.incr(f"views:{date.isoformat()}:{post_id}")

token = secrets.token_hex(16)
if redis.set(f"lock:{res}", token, nx=True, ex=30):
    try: ...
    finally:
        # Lua CAS release
        ...

redis.geoadd("drivers", lng, lat, driver_id)
nearby = redis.geosearch("drivers", longitude=user_lng, latitude=user_lat, radius=5, unit="km")

def variant(user_id, exp):
    bucket = redis.hget(f"exp:{exp}", user_id)
    if bucket: return bucket
    bucket = "A" if hash(f"{exp}:{user_id}") % 2 == 0 else "B"
    redis.hset(f"exp:{exp}", user_id, bucket)
    return bucket

def feature_enabled(feature, user_id):
    val = redis.hget(f"feature:{feature}", "config")
    config = json.loads(val) if val else {}
    if config.get("disabled"): return False
    if user_id in config.get("allowlist", []): return True
    if user_id in config.get("denylist", []): return False
    return user_id % 100 < config.get("percent", 0)

# Set
if redis.sadd("seen:events", event_id) == 0:
    return  # duplicate

# Bloom (memory efficient)
if redis.execute_command("BF.EXISTS", "seen", event_id):
    return
redis.execute_command("BF.ADD", "seen", event_id)

def page(after=None, limit=20):
    items = redis.zrangebyscore("posts", "(" + (after or "-inf"), "+inf", start=0, num=limit)
    next_cursor = items[-1].score if items else None
    return items, next_cursor

def invalidate(key):
    redis.delete(key)
    redis.publish("invalidations", key)

# Other servers listen and clear local in-memory cache

redis.expire(f"user:{id}", 86400 * 30)  # purge after 30 days

def shorten(url):
    code = base62.encode(redis.incr("counter"))
    redis.set(f"url:{code}", url)
    return code

# Increment score with decay
redis.zincrby("trending", 1, item_id)
redis.expire("trending", 3600)
top = redis.zrevrange("trending", 0, 9)

# Index
for word in extract_words(doc):
    redis.sadd(f"idx:{word}", doc_id)

# Search
results = redis.sinter([f"idx:{w}" for w in query_words])