pip install redis[hiredis]

# hiredis is a C parser — 2-5× faster protocol parsing

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

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

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")

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)

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": "[email protected]", "subject": "Welcome"})

    await enqueue_task(r, queue, {"to": "[email protected]", "subject": "Reset password"})

    print(f"Queue length: {await r.llen(queue)}")

    task = await dequeue_task(r, queue, timeout=1)

    print(f"Processed: {task}")

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

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())