Mock Distributed Rate Limiter with Dict in Python

import time
import threading
from collections import defaultdict


class DistributedRateLimiter:
    """
    A mock distributed rate limiter using a dict with thread-safe access.
    Implements a token bucket algorithm per user.
    """

    def __init__(self, rate_per_second=5, burst_capacity=10):
        self.rate_per_second = rate_per_second
        self.burst_capacity = burst_capacity
        self._buckets = defaultdict(lambda: {"tokens": burst_capacity, "last_refill": time.time()})
        self._lock = threading.Lock()

    def allow_request(self, user_id):
        with self._lock:
            bucket = self._buckets[user_id]
            now = time.time()
            elapsed = now - bucket["last_refill"]
            bucket["tokens"] = min(
                self.burst_capacity,
                bucket["tokens"] + elapsed * self.rate_per_second
            )
            bucket["last_refill"] = now

            if bucket["tokens"] >= 1:
                bucket["tokens"] -= 1
                return True
            return False


if __name__ == "__main__":
    limiter = DistributedRateLimiter(rate_per_second=2, burst_capacity=3)

    user = "user_42"

    # Simulate burst of requests
    for _ in range(5):
        print(f"Request allowed: {limiter.allow_request(user)}")

    # Wait for bucket to refill
    time.sleep(1)

    print(f"After 1s wait, request allowed: {limiter.allow_request(user)}")
    print(f"Request allowed: {limiter.allow_request(user)}")