Last active
March 22, 2023 03:56
-
-
Save pc3sq/d970875be2e223c98ef7f5cfe19d3c36 to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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