1class RateLimitingStrategy(ABC):
2    @abstractmethod
3    def allow_request(self, user_id: str) -> bool:
4        pass

1class FixedWindowStrategy(RateLimitingStrategy):
2    def __init__(self, max_requests: int, window_size_in_seconds: int):
3        self.max_requests = max_requests
4        self.window_size_in_millis = window_size_in_seconds * 1000
5        self.user_request_map: Dict[str, 'UserRequestInfo'] = {}
6        self._lock = threading.Lock()
7
8    def allow_request(self, user_id: str) -> bool:
9        current_time = int(time.time() * 1000)
10        
11        with self._lock:
12            if user_id not in self.user_request_map:
13                self.user_request_map[user_id] = UserRequestInfo(current_time)
14
15            request_info = self.user_request_map[user_id]
16
17            with request_info.lock:
18                if current_time - request_info.window_start >= self.window_size_in_millis:
19                    request_info.reset(current_time)
20
21                if request_info.request_count < self.max_requests:
22                    request_info.request_count += 1
23                    return True
24                else:
25                    return False

1class TokenBucket:
2    def __init__(self, capacity: int, refill_rate_per_second: int, current_time_millis: int):
3        self.capacity = capacity
4        self.refill_rate_per_second = refill_rate_per_second
5        self.tokens = capacity
6        self.last_refill_timestamp = current_time_millis
7        self.lock = threading.Lock()
8
9    def refill(self, current_time: int):
10        elapsed_time = current_time - self.last_refill_timestamp
11        tokens_to_add = int((elapsed_time / 1000.0) * self.refill_rate_per_second)
12
13        if tokens_to_add > 0:
14            self.tokens = min(self.capacity, self.tokens + tokens_to_add)
15            self.last_refill_timestamp = current_time
16
17class TokenBucketRateLimiter(RateLimiter):
18    def __init__(self, capacity: int, refill_rate_per_second: int):
19        self.capacity = capacity
20        self.refill_rate_per_second = refill_rate_per_second
21        self.user_buckets: Dict[str, 'TokenBucket'] = {}
22        self._lock = threading.Lock()
23
24    def allow_request(self, user_id: str) -> bool:
25        current_time = int(time.time() * 1000)
26        
27        with self._lock:
28            if user_id not in self.user_buckets:
29                self.user_buckets[user_id] = TokenBucket(self.capacity, self.refill_rate_per_second, current_time)
30            
31            bucket = self.user_buckets[user_id]
32
33            with bucket.lock:
34                bucket.refill(current_time)
35                if bucket.tokens > 0:
36                    bucket.tokens -= 1
37                    return True
38                else:
39                    return False

1class RateLimiterService:
2    _instance = None
3    _lock = threading.Lock()
4
5    def __init__(self):
6        if RateLimiterService._instance is not None:
7            raise Exception("This class is a singleton!")
8        self.rate_limiting_strategy = None
9
10    @staticmethod
11    def get_instance():
12        if RateLimiterService._instance is None:
13            with RateLimiterService._lock:
14                if RateLimiterService._instance is None:
15                    RateLimiterService._instance = RateLimiterService()
16        return RateLimiterService._instance
17
18    def set_rate_limiter(self, rate_limiting_strategy: RateLimitingStrategy):
19        self.rate_limiting_strategy = rate_limiting_strategy
20
21    def handle_request(self, user_id: str):
22        if self.rate_limiting_strategy.allow_request(user_id):
23            print(f"Request from user {user_id} is allowed")
24        else:
25            print(f"Request from user {user_id} is rejected: Rate limit exceeded")

1class RateLimiterDemo:
2    @staticmethod
3    def main():
4        user_id = "user123"
5
6        print("=== Fixed Window Demo ===")
7        RateLimiterDemo.run_fixed_window_demo(user_id)
8
9        print("\n=== Token Bucket Demo ===")
10        RateLimiterDemo.run_token_bucket_demo(user_id)
11
12    @staticmethod
13    def run_fixed_window_demo(user_id: str):
14        max_requests = 5
15        window_seconds = 10
16
17        rate_limiter = FixedWindowRateLimiter(max_requests, window_seconds)
18        service = RateLimiterService.get_instance()
19        service.set_rate_limiter(rate_limiter)
20
21        with ThreadPoolExecutor(max_workers=3) as executor:
22            for i in range(10):
23                executor.submit(service.handle_request, user_id)
24                try:
25                    time.sleep(0.5)
26                except KeyboardInterrupt:
27                    break
28
29    @staticmethod
30    def run_token_bucket_demo(user_id: str):
31        capacity = 5
32        refill_rate = 1  # 1 token per second
33
34        token_bucket_limiter = TokenBucketRateLimiter(capacity, refill_rate)
35        service = RateLimiterService.get_instance()
36        service.set_rate_limiter(token_bucket_limiter)
37
38        with ThreadPoolExecutor(max_workers=2) as executor:
39            # Simulate 10 rapid requests
40            for i in range(10):
41                executor.submit(service.handle_request, user_id)
42                try:
43                    time.sleep(0.3)  # faster than refill rate
44                except KeyboardInterrupt:
45                    break
46
47
48if __name__ == "__main__":
49    RateLimiterDemo.main()