circuitbreaker.py

import time


# CircuitBreaker implementation ################################
class CircuitOpenError(Exception):
    pass


class CircuitBreaker:
    def __init__(self, http_client, error_threshold, delay):
        """
        :param http_client: the http client to be used to make the calls
        :param threshold: number of failed attempts before the state is changed to "Open"
        :param delay: delay in seconds between "Closed" and "Open" state
        """
        self.http_client = http_client
        self.error_threshold = error_threshold
        self.delay = delay

################################################################


class DummyHttpResponse:
    def __init__(self, status_code):
        self.status_code = status_code


class DummyHttpClient:
    def __init__(self, responses):
        self.responses = responses
        self.counter_calls = {}

    def do_request(self, url):
        self.counter_calls.setdefault(url, 0)
        self.counter_calls[url] += 1
        return DummyHttpResponse(status_code=self.responses.pop())

    def get(self, url, *args, **kwargs):
        return self.do_request(url)

    def post(self, url, *args, **kwargs):
        return self.do_request(url)

    def put(self, url, *args, **kwargs):
        return self.do_request(url)

    def delete(self, url, *args, **kwargs):
        return self.do_request(url)


def tests():
    # stack of responses (the last ones are the first to be popped)
    responses = [200, 202, 201, 200, 200, 200, 200, 500, 500, 502, 503, 500]

    delay = 10
    client = DummyHttpClient(responses)
    breaker = CircuitBreaker(http_client=client, error_threshold=5, delay=delay)

    # failures calls
    for i in range(0, 5):
        breaker.do_request("GET", "https://url")

    assert client.counter_calls["https://url"] == 5

    try:
        breaker.do_request("GET", "https://url")
    except CircuitOpenError:
        # The circuit is open and the request doesn't go through
        assert client.counter_calls["https://url"] == 5
    else:
        assert False

    # Wait the delay time, before opening again
    time.sleep(delay)

    breaker.do_request("GET", "https://url")
    assert client.counter_calls["https://url"] == 6

    # success calls
    for i in range(0, 6):
        breaker.do_request("POST", "https://url")

    assert client.counter_calls["https://url"] == 12

if __name__ == '__main__':
    tests()