Pythonredis_patterns.py
Copy
1
import redis
2
import json
3
import time
4
 
5
r = redis.Redis(host="localhost", port=6379, decode_responses=True)
6
 
7
# Pattern 1: Cache-aside with TTL
8
def get_user(user_id: str) -> dict:
9
    cache_key = f"user:{user_id}"
10
    cached = r.get(cache_key)
11
    if cached:
12
        return json.loads(cached)
13
    # Simulate DB call
14
    user = {"id": user_id, "name": f"User {user_id}", "email": f"{user_id}@example.com"}
15
    r.setex(cache_key, 300, json.dumps(user))  # TTL: 5 minutes
16
    return user
17
 
18
print("Cache miss:", get_user("42"))
19
print("Cache hit:", get_user("42"))  # from cache
20
 
21
# Pattern 2: Sliding window rate limiter
22
def is_allowed(user_id: str, max_requests: int = 5, window_seconds: int = 60) -> bool:
23
    key = f"ratelimit:{user_id}"
24
    now = time.time()
25
    window_start = now - window_seconds
26
    with r.pipeline() as pipe:
27
        pipe.zremrangebyscore(key, 0, window_start)
28
        pipe.zadd(key, {str(now): now})
29
        pipe.zcard(key)
30
        pipe.expire(key, window_seconds)
31
        _, _, count, _ = pipe.execute()
32
    return count <= max_requests
33
 
34
for i in range(7):
35
    allowed = is_allowed("user-1")
36
    print(f"Request {i+1}: {'✓ allowed' if allowed else '✗ rate limited'}")
37
 
38
# Pattern 3: Leaderboard with sorted sets
39
r.delete("leaderboard")
40
scores = [("Alice", 1250), ("Bob", 980), ("Carol", 1500), ("Dave", 750), ("Eve", 1100)]
41
for name, score in scores:
42
    r.zadd("leaderboard", {name: score})
43
 
44
print("\nTop 3 leaderboard:")
45
top3 = r.zrevrange("leaderboard", 0, 2, withscores=True)
46
for rank, (name, score) in enumerate(top3, 1):
47
    print(f"  #{rank} {name}: {int(score)} pts")
48
 
49
# Pattern 4: Atomic counter
50
r.delete("page_views")
51
for _ in range(100):
52
    r.incr("page_views")
53
print(f"\nPage views: {r.get('page_views')}")
54
r.expire("page_views", 86400)  # Reset daily
55