Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
class ThrottleTransport:
def __init__(self, throttle, **kwargs):
self.pool = httpcore.SyncConnectionPool(**kwargs)
self.history = []
# Parse the thottle, which should be a string, like '100/minute'.
count, _, duration = throttle.partition('/')
self.max_in_history = int(count)
self.cutoff = {'second': 1.0, 'minute': 60.0, 'hour': 3600.0}[duration]
def request(method, url, headers, stream, timeout):
now = time.time()
while len(self.history) >= self.max_in_history:
expiry = now - self.cutoff
# Expiry old entries in the history.
self.history = [timestamp for timestamp in self.history if timestamp > expiry]
# Sleep for a bit if we've exceeded the throttle rate.
if len(self.history) >= self.max_in_history:
now = time.time()
return self.pool.request(method, url, headers, stream, timeout)
def close(self):
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment