Skip to content

Instantly share code, notes, and snippets.

@pc3sq
Last active March 22, 2023 03:56
Show Gist options
  • Save pc3sq/d970875be2e223c98ef7f5cfe19d3c36 to your computer and use it in GitHub Desktop.
Save pc3sq/d970875be2e223c98ef7f5cfe19d3c36 to your computer and use it in GitHub Desktop.
import time
from typing import List
class RateLimiter(object):
"""Given a maximum number of requests that can be made w/in
a given time interval in milliseconds, returns True if
you are permitted to make a request w/o being rate limited
or False when you are not permitted to make a request.
"""
def __init__(self, max_num_requests: int, interval: int, request_ts: List[int] = []):
self.max_num_requests = max_num_requests
self.interval = interval
self.request_ts = request_ts
def __str__(self):
return f"max_num_reqs: {self.max_num_requests}\ninterval (ms): {self.interval}\nearliest request: {self.request_ts[0]}\nlatest request: {self.request_ts[-1]}"
def can_request(self):
now = time.time()
if len(self.request_ts) < self.max_num_requests:
self.request_ts.append(now)
return True
if self.request_ts[0] < now - self.interval:
self.request_ts.pop(0)
self.request_ts.append(now)
return True
return False
if __name__ == "__main__":
# instantiate a rate_limiter
params_1 = (10, 10, [time.time() - (n * 1000) for n in range(10)])
limiter_1 = RateLimiter(*params_1)
print(limiter_1)
# should NOT be able to make an immediate request
print(f"Request immediately: {limiter_1.can_request()}")
# after waiting 10 secs
print('Sleeping 10 secs ...')
time.sleep(10)
# should now be able to request 10x immediately
for i in range(10):
print(f"Request {i+1}: {limiter_1.can_request()}")
# but not the 11th time immediately
print(f"Request 11: {limiter_1.can_request()}")
# after waiting 10 secs
print('Sleeping 10 secs ...')
time.sleep(10)
# make another 10x requests with 1 sec in b/t each request
for i in range(10):
time.sleep(1)
print(f"Request {i+1}: {limiter_1.can_request()}")
# can't make an immediate 11th request
print(f"Request 11: {limiter_1.can_request()}")
# but can after waiting a second
time.sleep(1)
print(f"Request 12 after pausing 1 sec: {limiter_1.can_request()}")
# and in fact if I keep making requests at 1 sec intervals, I can go in
# perpetuity b/c it is equal to the rate at which I make requests w/o
# limitation, 10 requests every 10 secs = 1 req/sec
for i in range(30):
time.sleep(1)
print(f"Request {i+1}: {limiter_1.can_request()}")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment