1. import time
  2. import threading
  4. class RequestThrottler:
  5. def __init__(self, rate_limit_calls, period):
  6. """
  7. Initializes the throttler.
  9. Args:
  10. rate_limit_calls (int): Maximum number of requests allowed within the period.
  11. period (float): Time period (in seconds) for the rate limit.
  12. """
  13. self.rate_limit_calls = rate_limit_calls
  14. self.period = period
  15. self.request_times = [] # Store timestamps of requests
  16. self.lock = threading.Lock() #For thread safety
  18. def is_allowed(self):
  19. """
  20. Checks if a request is allowed based on the rate limit.
  21. Returns True if allowed, False otherwise.
  22. """
  23. with self.lock:
  24. now = time.time()
  25. # Remove requests older than the period
  26. self.request_times = [t for t in self.request_times if now - t < self.period]
  28. if len(self.request_times) < self.rate_limit_calls:
  29. self.request_times.append(now)
  30. return True
  31. else:
  32. return False
  34. def wait_if_needed(self):
  35. """
  36. Waits if the request is not allowed and then checks again.
  37. """
  38. if not self.is_allowed():
  39. time.sleep(self.period / self.rate_limit_calls) # Calculate wait time
  40. self.is_allowed() # check again after waiting

Add your comment