Code Generation

Implementation: rate_limiter.py

Approach

Token bucket rate limiter with Redis backend for distributed use. Chose token bucket over sliding window for burst tolerance — allows short traffic spikes while enforcing a sustained rate ceiling. Redis MULTI/EXEC ensures atomicity across concurrent requests. Includes configurable bucket size and refill rate per client key.

Code

import time
import redis

class TokenBucketRateLimiter:
    def __init__(self, redis_client: redis.Redis, max_tokens: int = 100, refill_rate: float = 10.0):
        self.redis = redis_client
        self.max_tokens = max_tokens
        self.refill_rate = refill_rate  # tokens per second

    def allow_request(self, client_key: str) -> bool:
        key = f"ratelimit:{client_key}"
        now = time.time()
        pipe = self.redis.pipeline()
        pipe.hmget(key, "tokens", "last_refill")
        result = pipe.execute()[0]
        tokens = float(result[0]) if result[0] else self.max_tokens
        last_refill = float(result[1]) if result[1] else now
        elapsed = now - last_refill
        tokens = min(self.max_tokens, tokens + elapsed * self.refill_rate)
        if tokens >= 1:
            tokens -= 1
            allowed = True
        else:
            allowed = False
        pipe.hset(key, mapping={"tokens": tokens, "last_refill": now})
        pipe.expire(key, int(self.max_tokens / self.refill_rate) + 60)
        pipe.execute()
        return allowed

Tests

def test_allows_requests_within_limit():
    limiter = TokenBucketRateLimiter(redis_client, max_tokens=5, refill_rate=1.0)
    results = [limiter.allow_request("test-client") for _ in range(5)]
    assert all(results)  # First 5 requests allowed

def test_blocks_after_exhaustion():
    limiter = TokenBucketRateLimiter(redis_client, max_tokens=2, refill_rate=0.1)
    limiter.allow_request("exhaust-client")
    limiter.allow_request("exhaust-client")
    assert not limiter.allow_request("exhaust-client")  # Third request blocked

Execution Output

test_allows_requests_within_limit PASSED
test_blocks_after_exhaustion PASSED
2 passed in 0.34s