Installation

deckar01-ratelimit

$ pip install -r requirements.txt

$ pip install deckar01-ratelimit

$ git clone https://github.com/deckar01/ratelimit
$ cd ratelimit
$ python setup.py install

Usage

To use this package simply decorate any function that makes an API call:

from ratelimit import limits

import requests

FIFTEEN_MINUTES = 900

@limits(calls=15, period=FIFTEEN_MINUTES)
def call_api(url):
    response = requests.get(url)

    if response.status_code != 200:
        raise Exception('API response: {}'.format(response.status_code))
    return response

This function will not be able to make more then 15 API call within a 15 minute time period.

The arguments passed into the decorator describe the number of function invocation allowed over a specified time period (in seconds). If no time period is specified then it defaults to 15 minutes (the time window imposed by Twitter).

If a decorated function is called more times than that allowed within the specified time period then a ratelimit.RateLimitException is raised. This may be used to implement a retry strategy such as an expoential backoff

from ratelimit import limits, RateLimitException
from backoff import on_exception, expo

import requests

FIFTEEN_MINUTES = 900

@on_exception(expo, RateLimitException, max_tries=8)
@limits(calls=15, period=FIFTEEN_MINUTES)
def call_api(url):
    response = requests.get(url)

    if response.status_code != 200:
        raise Exception('API response: {}'.format(response.status_code))
    return response

Alternatively to cause the current thread to sleep until the specified time period has ellapsed and then retry the function use the sleep_and_retry decorator. This ensures that every function invocation is successful at the cost of halting the thread.

from ratelimit import limits, sleep_and_retry

import requests

FIFTEEN_MINUTES = 900

@sleep_and_retry
@limits(calls=15, period=FIFTEEN_MINUTES)
def call_api(url):
    response = requests.get(url)

    if response.status_code != 200:
        raise Exception('API response: {}'.format(response.status_code))
    return response

from ratelimit import limits, sleep_and_retry

import requests

FIFTEEN_MINUTES = 900

@sleep_and_retry
@limits(calls=15, period=FIFTEEN_MINUTES, storage='ratelimit.db')
def call_api(url):
    response = requests.get(url)

    if response.status_code != 200:
        raise Exception('API response: {}'.format(response.status_code))
    return response

from ratelimit import limits, sleep_and_retry

import requests

HOUR = 3600
DAY = 24*HOUR

@sleep_and_retry
@limits(calls=15, period=HOUR, storage='ratelimit.db', name='hourly_limit')
@sleep_and_retry
@limits(calls=100, period=DAY, storage='ratelimit.db', name='daily_limit')
def call_api(url):
    response = requests.get(url)

    if response.status_code != 200:
        raise Exception('API response: {}'.format(response.status_code))
    return response

License

This project is licensed under the MIT License.