A simple circuitbreaker, useful for timeouts, etc

circuitbreaker.py

from __future__ import print_function  # I wrote this in py2, but wanted py3 compatibility
from time import sleep
from datetime import datetime, timedelta


class TrippedBreaker(Exception):
    pass


class CircuitBreaker(object):
    """
    Raise TrippedBreaker if more than x exceptions counted in y interval
    """
    def __init__(self, threshold, interval, units='minutes'):
        self.threshold = threshold
        kinterval = {units: int(interval)}  # TODO: Don't assume people use this right
        self.interval = timedelta(**kinterval)
        self._history = []

    @property
    def threshold_reached(self):
        """
        Return True if we have reached our threshold
        """
        return len(self._history) > self.threshold

    def increment(self):
        self._history.append(datetime.utcnow())
        if not self.threshold_reached:
            pass  # No need to take action
        elif self._history[0] >= datetime.utcnow() - self.interval:
            raise TrippedBreaker
        else:
            while self.threshold_reached:
                self._history.pop(0)  # quick cleanup task


if __name__ == "__main__":
    breaker = CircuitBreaker(5, 10, units='seconds')
    try:
        for i in range(10):
            sleep(0.1)
            try:
                raise Exception("I am a random exception")
            except Exception as e:
                print(e)
                breaker.increment()
    except TrippedBreaker:
        print("Got tripped breaker!")
    else:
        print("Didn't trip breaker")