Skip to content

Instantly share code, notes, and snippets.

@observerss
Created October 12, 2013 03:24
Show Gist options
  • Save observerss/6945419 to your computer and use it in GitHub Desktop.
Save observerss/6945419 to your computer and use it in GitHub Desktop.
Python snippet that throttles a function call. This is useful when the function request a remote service that having throttling policy. We must honor the throttling, otherwise we will be banned.
import time
from collections import deque
def throttled_call(func, args=(), kwargs={}, threshold_per_minute=60):
"""Throttling the function call with ``threshold_per_minute`` calls per minute.
This is useful when the function request a remote service that having throttling policy.
We must honor the throttling, otherwise we will be banned.
:param func: the function to be called
:param args: args of that function
:param kwargs: kwargs of that function
:param threshold_per_minute: defines how many calls can be made to the function per minute
Example Usage::
>>> import time
>>> def print_hello():
... print 'hello'
and calling the following will make the ``print_hello`` function less annoying
>>> while True:
>>> throttled_call(print_hello, threshold_per_minute=60)
"""
if not hasattr(throttled_call, 'logs'):
throttled_call.logs = deque()
logs = throttled_call.logs
def remove_outdated():
# remove outdated from logs
t = time.time()
while True:
if logs and logs[0] < t - 60:
logs.popleft()
else:
break
def wait_for_threshold():
while len(logs) > threshold_per_minute:
remove_outdated()
time.sleep(0.3)
def smoothen_calling_interval():
average_processing_time = (time.time() - logs[0]) / len(logs)
expected_processing_time = 60. / threshold_per_minute
if expected_processing_time > average_processing_time:
time.sleep((len(logs)+0.8)*expected_processing_time - len(logs)*average_processing_time)
if logs and len(logs) < threshold_per_minute:
smoothen_calling_interval()
else:
wait_for_threshold()
logs.append(time.time())
return func(*args, **kwargs)
def test_hello():
def print_hello():
print 'hello'
while True:
throttled_call(print_hello, threshold_per_minute=60)
if __name__ == '__main__':
test_hello()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment