Python Redis: Caching and Fast Data Structures

Redis is an in-memory data store used for caching, session storage, pub/sub messaging, leaderboards, rate limiting, and more. With redis-py's async client, it integrates cleanly into any asyncio application.

Installation

pip install redis[hiredis]
# hiredis is a C parser — 2-5× faster protocol parsing

Connect and Verify

import asyncio
import redis.asyncio as aioredis
from datetime import timedelta
import json

REDIS_URL = "redis://localhost:6379/0"

async def get_redis() -> aioredis.Redis:
    client = aioredis.from_url(
        REDIS_URL,
        encoding="utf-8",
        decode_responses=True,
        socket_connect_timeout=5,
        socket_timeout=5,
        retry_on_timeout=True,
    )
    pong = await client.ping()
    print(f"Redis connected: {pong}")
    return client

Strings — Basic Cache with TTL

async def cache_set(r: aioredis.Redis, key: str, value: str, ttl: int = 300) -> None:
    await r.set(key, value, ex=ttl)

async def cache_get(r: aioredis.Redis, key: str) -> str | None:
    return await r.get(key)

# Cache-aside pattern
async def get_user_profile(r: aioredis.Redis, user_id: int, db) -> dict:
    cache_key = f"user:profile:{user_id}"
    cached = await r.get(cache_key)
    if cached:
        print(f"Cache HIT for user {user_id}")
        return json.loads(cached)

    print(f"Cache MISS for user {user_id} — querying DB")
    user = await db.fetch_user(user_id)   # your DB call
    if user:
        await r.set(cache_key, json.dumps(user), ex=600)
    return user or {}

# Atomic counter
async def increment_page_views(r: aioredis.Redis, page: str) -> int:
    key = f"views:{page}"
    count = await r.incr(key)
    await r.expire(key, 86400)   # reset counter after 24 h
    return count

Hashes — Structured Objects

async def save_session(r: aioredis.Redis, session_id: str, data: dict, ttl: int = 3600) -> None:
    key = f"session:{session_id}"
    await r.hset(key, mapping=data)
    await r.expire(key, ttl)

async def get_session(r: aioredis.Redis, session_id: str) -> dict:
    return await r.hgetall(f"session:{session_id}")

async def update_session_field(r: aioredis.Redis, session_id: str, field: str, value: str) -> None:
    await r.hset(f"session:{session_id}", field, value)

async def demo_sessions(r: aioredis.Redis):
    await save_session(r, "abc123", {"user_id": "42", "role": "admin", "lang": "en"})
    session = await get_session(r, "abc123")
    print(f"Session: {session}")
    await update_session_field(r, "abc123", "lang", "fr")

Sorted Sets — Leaderboards

async def add_score(r: aioredis.Redis, board: str, player: str, score: float) -> None:
    await r.zadd(board, {player: score})

async def get_top_players(r: aioredis.Redis, board: str, n: int = 10) -> list[tuple]:
    return await r.zrevrange(board, 0, n - 1, withscores=True)

async def get_player_rank(r: aioredis.Redis, board: str, player: str) -> int | None:
    rank = await r.zrevrank(board, player)
    return rank + 1 if rank is not None else None

async def demo_leaderboard(r: aioredis.Redis):
    board = "game:weekly"
    players = [("alice", 9800), ("bob", 7200), ("carol", 11500), ("dave", 8900)]
    for name, score in players:
        await add_score(r, board, name, score)

    top = await get_top_players(r, board, 3)
    print("Top 3 players:")
    for i, (player, score) in enumerate(top, 1):
        print(f"  #{i} {player}: {int(score)}")
    await r.delete(board)

Lists — Simple Task Queue

async def enqueue_task(r: aioredis.Redis, queue: str, task: dict) -> int:
    return await r.lpush(queue, json.dumps(task))

async def dequeue_task(r: aioredis.Redis, queue: str, timeout: int = 5) -> dict | None:
    result = await r.brpop(queue, timeout=timeout)
    if result:
        _, raw = result
        return json.loads(raw)
    return None

async def demo_queue(r: aioredis.Redis):
    queue = "tasks:email"
    await enqueue_task(r, queue, {"to": "[e-mail]", "subject": "Welcome"})
    await enqueue_task(r, queue, {"to": "[e-mail]", "subject": "Reset password"})
    print(f"Queue length: {await r.llen(queue)}")
    task = await dequeue_task(r, queue, timeout=1)
    print(f"Processed: {task}")

Rate Limiting with Sliding Window

async def is_rate_limited(
    r: aioredis.Redis,
    identifier: str,
    limit: int = 100,
    window: int = 60,
) -> bool:
    key = f"ratelimit:{identifier}"
    pipe = r.pipeline()
    now = asyncio.get_event_loop().time()
    window_start = now - window

    pipe.zremrangebyscore(key, 0, window_start)
    pipe.zadd(key, {str(now): now})
    pipe.zcard(key)
    pipe.expire(key, window)
    results = await pipe.execute()

    current_count = results[2]
    return current_count > limit

Full Working Example

async def main():
    r = await get_redis()

    await cache_set(r, "greeting", "Hello, Redis!", ttl=60)
    print(await cache_get(r, "greeting"))

    await demo_sessions(r)
    await demo_leaderboard(r)
    await demo_queue(r)

    limited = await is_rate_limited(r, "user:42", limit=5, window=60)
    print(f"Rate limited: {limited}")

    await r.aclose()

if __name__ == "__main__":
    asyncio.run(main())

Practical Tips

• Always set TTLs on cached keys — unbounded growth will exhaust memory
• Use pipeline() to batch multiple commands into one round-trip
• hiredis provides a C extension parser — always install it in production
• decode_responses=True returns strings instead of bytes — simpler code
• Sorted sets are O(log N) for add/remove — ideal for leaderboards and timelines
• brpop/blpop block until an item arrives — perfect for worker queues

---

Follow me for more Python tips! 🐍

---

💡 Related: **Content Creator Ultimate Bundle (Save 33%)* — $29.99*