Simple Thread Safe Python Rate Limiter. MIT License. Intended for use in e.g. limiting calls to rate limited services.

rate_limiter.py

# MIT License
# author @jdmaturen

import logging
import time
from collections import deque
from threading import Lock


class RateLimiter(object):
    """
    Simple rate limiter. Returns the number of seconds to sleep if any when called.

    >>> r = RateLimiter(calls_per_period=1, period_size=0.1)
    >>> r(); r(); r()
    0.0
    0.09994101524353027
    0.19990897178649902
    """

    def __init__(self, calls_per_period, period_size=1.0):
        self.calls_per_period = calls_per_period
        self.period_size = float(period_size)

        self.calls = deque()
        self.lock = Lock()

    def __call__(self, *args, **kwargs):
        """
        return the number of seconds, if any, that should be slept by the caller

        :param args:
        :param kwargs:
        :return: float seconds for caller to sleep to stay in rate limit
        """
        with self.lock:
            t0 = time.time()

            # clean up stale data
            while len(self.calls) > 0 and (t0 - self.calls[0]) > self.period_size:
                self.calls.popleft()

            dt = 0.0
            if len(self.calls) >= self.calls_per_period:
                until = self.calls[len(self.calls) - self.calls_per_period] + self.period_size
                dt = max(0.0, until - t0)

            # append t0 + dt to the list as we want future calls to reference
            # when this action _would_ take place, not when it was called
            self.calls.append(t0 + dt)
            return dt