Python decorator for throttling and ramping function calls.

throttle.py

from datetime import datetime, timedelta
from functools import wraps

class throttle(object):
    """
    Decorator that prevents a function from being called more than once every
    time period.

    To create a function that cannot be called more than once a minute:

        @throttle(minutes=1)
        def my_fun():
            pass
    """

    def __init__(self, seconds=0, minutes=0, hours=0):
        self.throttle_period = timedelta(
            seconds=seconds, minutes=minutes, hours=hours
        )
        self.time_of_last_call = datetime.min

    def __call__(self, fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            now = datetime.now()
            time_since_last_call = now - self.time_of_last_call

            if time_since_last_call > self.throttle_period:
                self.time_of_last_call = now
                return fn(*args, **kwargs)

        return wrapper


class multi_throttle(object):
    """Forked from https://gist.github.com/ChrisTM/5834503"""
    # FIXME: needs testing

    def __init__(self, seconds=0, minutes=0, hours=0, minimum=None, blocking=False):
        self.throttle_period = timedelta(seconds=seconds, minutes=minutes, hours=hours) if (seconds or minutes or hours) else None
        self.minimum_throttle_period = timedelta(**minimum) if minimum else None
        self.time_of_last_call = datetime.min
        assert not blocking

    def __call__(self, fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            now = datetime.now()
            time_since_last_call = now - self.time_of_last_call

            # if self.blocking and self.throttle_period:
            #     time_left = self.throttle_period - time_since_last_call
            #     if time_left > timedelta(seconds=0):
            #         time.sleep(time_left.seconds)
            if (self.throttle_period and time_since_last_call > self.throttle_period) or (self.minimum_throttle_period and time_since_last_call < self.minimum_throttle_period):
                self.time_of_last_call = now
                return fn(*args, **kwargs)

        return wrapper


class ramp_single(object):
    # FIXME: needs testing

    def __init__(self, seconds=0, minutes=0, hours=0):
        self.ramp_period = timedelta(
            seconds=seconds, minutes=minutes, hours=hours
        )
        self.time_of_last_call = datetime.min

    def __call__(self, fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            now = datetime.now()
            time_since_last_call = now - self.time_of_last_call

            if time_since_last_call > self.ramp_period:
                self.time_of_last_call = now
                return fn(*args, **kwargs)

        return wrapper


class ramp(object):
    """
    Decorator that executes a function if it's called at least count times every time period.
    """

    def __init__(self, count=2, seconds=0, minutes=0, hours=0):
        self.times = collections.deque(maxlen=count)
        self.ramp_period = timedelta(seconds=seconds, minutes=minutes, hours=hours)

    def __call__(self, fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            now = datetime.now()
            self.times.append(now)
            while self.times and self.times[0] < (now - self.ramp_period):
                self.times.popleft()
            if len(self.times) == self.times.maxlen:
                return fn(*args, **kwargs)

        return wrapper


"""
@mode.throttle(minutes=1)
def ringer():
    print "ringer()"
@mode.ramp(3, minutes=1)
def ringer():
    print "ringer()"
@mode.ramp(3, minutes=5)
@mode.throttle(minutes=1)
def ringer():
    print "ringer()"
"""