Python Async Redis Sliding Window Rate Limiter

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import time
import redis.asyncio as redis

async def check_rate_limit(r: redis.Redis, user_id: str, limit: int, window_sec: int) -> bool:
    now = time.time()
    key = f"rate:{user_id}"
    window_start = now - window_sec

    async with r.pipeline(transaction=True) as pipe:
        pipe.zremrangebyscore(key, 0, window_start)
        pipe.zadd(key, {str(now): now})
        pipe.zcard(key)
        pipe.expire(key, window_sec)
        results = await pipe.execute()

    current_requests = results[2]
    return current_requests <= limit