circuitbreaker.CircuitBreaker.closed() - Code Metrics - Inspection of "Use a monotonic clock source" - fabfuel/circuitbreaker - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — develop (#31)

by Thomas

created 2022-01-12 18:11 UTC

circuitbreaker.CircuitBreaker.closed() A

↳ Parent: circuitbreaker

Complexity

Conditions

Size

Total Lines	3
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	2

Importance

Changes

Metric	Value
cc	1
eloc	3
nop	1
dl	0
loc	3
ccs	0
cts	2
cp	0
crap	2
rs	10
c	0
b	0
f	0

# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from __future__ import division
from __future__ import print_function
from __future__ import absolute_import

from functools import wraps
from inspect import isgeneratorfunction
from typing import AnyStr, Iterable
from time import monotonic

STATE_CLOSED = 'closed'
STATE_OPEN = 'open'
STATE_HALF_OPEN = 'half_open'


class CircuitBreaker(object):
    FAILURE_THRESHOLD = 5
    RECOVERY_TIMEOUT = 30
    EXPECTED_EXCEPTION = Exception
    FALLBACK_FUNCTION = None

    def __init__(self,
                 failure_threshold=None,
                 recovery_timeout=None,
                 expected_exception=None,
                 name=None,
                 fallback_function=None):
        self._last_failure = None
        self._failure_count = 0
        self._failure_threshold = failure_threshold or self.FAILURE_THRESHOLD
        self._recovery_timeout = recovery_timeout or self.RECOVERY_TIMEOUT
        self._expected_exception = expected_exception or self.EXPECTED_EXCEPTION
        self._fallback_function = fallback_function or self.FALLBACK_FUNCTION
        self._name = name
        self._state = STATE_CLOSED
        self._opened = monotonic()

    def __call__(self, wrapped):
        return self.decorate(wrapped)

    def __enter__(self):
        return None

    def __exit__(self, exc_type, exc_value, _traceback):
        if exc_type and issubclass(exc_type, self._expected_exception):
            # exception was raised and is our concern
            self._last_failure = exc_value
            self.__call_failed()
        else:
            self.__call_succeeded()
        return False  # return False to raise exception if any

    def decorate(self, function):
        """
        Applies the circuit breaker to a function
        """
        if self._name is None:
            self._name = function.__name__

        CircuitBreakerMonitor.register(self)

        if isgeneratorfunction(function):
            call = self.call_generator
        else:
            call = self.call

        @wraps(function)
        def wrapper(*args, **kwargs):
            if self.opened:
                if self.fallback_function:
                    return self.fallback_function(*args, **kwargs)
                raise CircuitBreakerError(self)
            return call(function, *args, **kwargs)

        return wrapper

    def call(self, func, *args, **kwargs):
        """
        Calls the decorated function and applies the circuit breaker
        rules on success or failure
        :param func: Decorated function
        """
        with self:
            return func(*args, **kwargs)

    def call_generator(self, func, *args, **kwargs):
        """
        Calls the decorated generator function and applies the circuit breaker
        rules on success or failure
        :param func: Decorated generator function
        """
        with self:
            for el in func(*args, **kwargs):
                yield el

    def __call_succeeded(self):
        """
        Close circuit after successful execution and reset failure count
        """
        self._state = STATE_CLOSED
        self._last_failure = None
        self._failure_count = 0

    def __call_failed(self):
        """
        Count failure and open circuit, if threshold has been reached
        """
        self._failure_count += 1
        if self._failure_count >= self._failure_threshold:
            self._state = STATE_OPEN
            self._opened = monotonic()

    @property
    def state(self):
        if self._state == STATE_OPEN and self.open_remaining <= 0:
            return STATE_HALF_OPEN
        return self._state

    @property
    def open_until(self):
        """
        The monotime when the circuit breaker will try to recover
        :return: float
        """
        return self._opened + self._recovery_timeout

    @property
    def open_remaining(self):
        """
        Number of seconds remaining, the circuit breaker stays in OPEN state
        :return: float
        """
        return self.open_until - monotonic()

    @property
    def failure_count(self):
        return self._failure_count

    @property
    def closed(self):
        return self.state == STATE_CLOSED

    @property
    def opened(self):
        return self.state == STATE_OPEN

    @property
    def name(self):
        return self._name

    @property
    def last_failure(self):
        return self._last_failure

    @property
    def fallback_function(self):
        return self._fallback_function

    def __str__(self, *args, **kwargs):
        return self._name


class CircuitBreakerError(Exception):
    def __init__(self, circuit_breaker, *args, **kwargs):
        """
        :param circuit_breaker:
        :param args:
        :param kwargs:
        :return:
        """
        super(CircuitBreakerError, self).__init__(*args, **kwargs)
        self._circuit_breaker = circuit_breaker

    def __str__(self, *args, **kwargs):
        return 'Circuit "%s" OPEN until %s (%d failures, %d sec remaining) (last_failure: %r)' % (
            self._circuit_breaker.name,
            self._circuit_breaker.open_until,
            self._circuit_breaker.failure_count,
            round(self._circuit_breaker.open_remaining),
            self._circuit_breaker.last_failure,
        )


class CircuitBreakerMonitor(object):
    circuit_breakers = {}

    @classmethod
    def register(cls, circuit_breaker):
        cls.circuit_breakers[circuit_breaker.name] = circuit_breaker

    @classmethod
    def all_closed(cls):
        # type: () -> bool
        return len(list(cls.get_open())) == 0

    @classmethod
    def get_circuits(cls):
        # type: () -> Iterable[CircuitBreaker]
        return cls.circuit_breakers.values()

    @classmethod
    def get(cls, name):
        # type: (AnyStr) -> CircuitBreaker
        return cls.circuit_breakers.get(name)

    @classmethod
    def get_open(cls):
        # type: () -> Iterable[CircuitBreaker]
        for circuit in cls.get_circuits():
            if circuit.opened:
                yield circuit

    @classmethod
    def get_closed(cls):
        # type: () -> Iterable[CircuitBreaker]
        for circuit in cls.get_circuits():
            if circuit.closed:
                yield circuit


def circuit(failure_threshold=None,
            recovery_timeout=None,
            expected_exception=None,
            name=None,
            fallback_function=None,
            cls=CircuitBreaker):

    # if the decorator is used without parameters, the
    # wrapped function is provided as first argument
    if callable(failure_threshold):
        return cls().decorate(failure_threshold)
    else:
        return cls(
            failure_threshold=failure_threshold,
            recovery_timeout=recovery_timeout,
            expected_exception=expected_exception,
            name=name,
            fallback_function=fallback_function)


1		# -- coding: utf-8 --
2	1	from __future__ import unicode_literals
3	1	from __future__ import division
4	1	from __future__ import print_function
5	1	from __future__ import absolute_import
6
7	1	from functools import wraps
8	1	from inspect import isgeneratorfunction
9	1	from typing import AnyStr, Iterable
10	1	from time import monotonic
11
12		STATE_CLOSED = 'closed'
13		STATE_OPEN = 'open'
14		STATE_HALF_OPEN = 'half_open'
15
16
17		class CircuitBreaker(object):
18		FAILURE_THRESHOLD = 5
19		RECOVERY_TIMEOUT = 30
20		EXPECTED_EXCEPTION = Exception
21		FALLBACK_FUNCTION = None
22
23		def __init__(self,
24		failure_threshold=None,
25		recovery_timeout=None,
26		expected_exception=None,
27		name=None,
28		fallback_function=None):
29		self._last_failure = None
30		self._failure_count = 0
31		self._failure_threshold = failure_threshold or self.FAILURE_THRESHOLD
32		self._recovery_timeout = recovery_timeout or self.RECOVERY_TIMEOUT
33		self._expected_exception = expected_exception or self.EXPECTED_EXCEPTION
34		self._fallback_function = fallback_function or self.FALLBACK_FUNCTION
35		self._name = name
36		self._state = STATE_CLOSED
37		self._opened = monotonic()
38
39		def __call__(self, wrapped):
40		return self.decorate(wrapped)
41
42		def __enter__(self):
43		return None
44
45		def __exit__(self, exc_type, exc_value, _traceback):
46		if exc_type and issubclass(exc_type, self._expected_exception):
47		# exception was raised and is our concern
48		self._last_failure = exc_value
49		self.__call_failed()
50		else:
51		self.__call_succeeded()
52		return False # return False to raise exception if any
53
54		def decorate(self, function):
55		"""
56		Applies the circuit breaker to a function
57		"""
58		if self._name is None:
59		self._name = function.__name__
60
61		CircuitBreakerMonitor.register(self)
62
63		if isgeneratorfunction(function):
64		call = self.call_generator
65		else:
66		call = self.call
67
68		@wraps(function)
69		def wrapper(args, *kwargs):
70		if self.opened:
71		if self.fallback_function:
72		return self.fallback_function(args, *kwargs)
73		raise CircuitBreakerError(self)
74		return call(function, args, *kwargs)
75
76		return wrapper
77
78		def call(self, func, args, *kwargs):
79		"""
80		Calls the decorated function and applies the circuit breaker
81		rules on success or failure
82		:param func: Decorated function
83		"""
84		with self:
85		return func(args, *kwargs)
86
87		def call_generator(self, func, args, *kwargs):
88		"""
89		Calls the decorated generator function and applies the circuit breaker
90		rules on success or failure
91		:param func: Decorated generator function
92		"""
93		with self:
94		for el in func(args, *kwargs):
95		yield el
96
97		def __call_succeeded(self):
98		"""
99		Close circuit after successful execution and reset failure count
100		"""
101		self._state = STATE_CLOSED
102		self._last_failure = None
103		self._failure_count = 0
104
105		def __call_failed(self):
106		"""
107		Count failure and open circuit, if threshold has been reached
108		"""
109		self._failure_count += 1
110		if self._failure_count >= self._failure_threshold:
111		self._state = STATE_OPEN
112		self._opened = monotonic()
113
114		@property
115		def state(self):
116		if self._state == STATE_OPEN and self.open_remaining <= 0:
117		return STATE_HALF_OPEN
118		return self._state
119
120		@property
121		def open_until(self):
122		"""
123		The monotime when the circuit breaker will try to recover
124		:return: float
125		"""
126		return self._opened + self._recovery_timeout
127
128		@property
129		def open_remaining(self):
130		"""
131		Number of seconds remaining, the circuit breaker stays in OPEN state
132		:return: float
133		"""
134		return self.open_until - monotonic()
135
136		@property
137		def failure_count(self):
138		return self._failure_count
139
140		@property
141		def closed(self):
142		return self.state == STATE_CLOSED
143
144		@property
145		def opened(self):
146		return self.state == STATE_OPEN
147
148		@property
149		def name(self):
150		return self._name
151
152		@property
153		def last_failure(self):
154		return self._last_failure
155
156		@property
157		def fallback_function(self):
158		return self._fallback_function
159
160		def __str__(self, args, *kwargs):
161		return self._name
162
163
164		class CircuitBreakerError(Exception):
165		def __init__(self, circuit_breaker, args, *kwargs):
166		"""
167		:param circuit_breaker:
168		:param args:
169		:param kwargs:
170		:return:
171		"""
172		super(CircuitBreakerError, self).__init__(args, *kwargs)
173		self._circuit_breaker = circuit_breaker
174
175		def __str__(self, args, *kwargs):
176		return 'Circuit "%s" OPEN until %s (%d failures, %d sec remaining) (last_failure: %r)' % (
177		self._circuit_breaker.name,
178		self._circuit_breaker.open_until,
179		self._circuit_breaker.failure_count,
180		round(self._circuit_breaker.open_remaining),
181		self._circuit_breaker.last_failure,
182		)
183
184
185		class CircuitBreakerMonitor(object):
186		circuit_breakers = {}
187
188		@classmethod
189		def register(cls, circuit_breaker):
190		cls.circuit_breakers[circuit_breaker.name] = circuit_breaker
191
192		@classmethod
193		def all_closed(cls):
194		# type: () -> bool
195		return len(list(cls.get_open())) == 0
196
197		@classmethod
198		def get_circuits(cls):
199		# type: () -> Iterable[CircuitBreaker]
200		return cls.circuit_breakers.values()
201
202		@classmethod
203		def get(cls, name):
204		# type: (AnyStr) -> CircuitBreaker
205		return cls.circuit_breakers.get(name)
206
207		@classmethod
208		def get_open(cls):
209		# type: () -> Iterable[CircuitBreaker]
210		for circuit in cls.get_circuits():
211		if circuit.opened:
212		yield circuit
213
214		@classmethod
215		def get_closed(cls):
216		# type: () -> Iterable[CircuitBreaker]
217		for circuit in cls.get_circuits():
218		if circuit.closed:
219		yield circuit
220
221
222		def circuit(failure_threshold=None,
223		recovery_timeout=None,
224		expected_exception=None,
225		name=None,
226		fallback_function=None,
227		cls=CircuitBreaker):
228
229		# if the decorator is used without parameters, the
230		# wrapped function is provided as first argument
231		if callable(failure_threshold):
232		return cls().decorate(failure_threshold)
233		else:
234		return cls(
235		failure_threshold=failure_threshold,
236		recovery_timeout=recovery_timeout,
237		expected_exception=expected_exception,
238		name=name,
239		fallback_function=fallback_function)
240

fabfuel / circuitbreaker

Pull Request — develop (#31)

circuitbreaker.CircuitBreaker.closed() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like