Code Generation

Implementation — rate_limiter.py

Design choices

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

test_allows_requests_within_limit PASSED
test_blocks_after_exhaustion PASSED
2 passed in 0.34s

Tests written this round

Follow-ups

This sample illustrates the skill's output format. The implementation is illustrative; harden before production use.