circuitbreaker.CircuitBreaker.last_failure() - Code Metrics - Inspection of "Merge branch 'release/1.2.0'" - fabfuel/circuitbreaker - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 6a4cde...f28c44 )

by Fabian

created 2019-06-30 03:54 UTC

circuitbreaker.CircuitBreaker.last_failure() A

↳ Parent: circuitbreaker

Complexity

Conditions

Size

Total Lines	3
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	3
nop	1
dl	0
loc	3
ccs	2
cts	2
cp	1
crap	1
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 datetime import datetime, timedelta
from typing import AnyStr, Iterable

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


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

    def __init__(self,
                 failure_threshold=None,
                 recovery_timeout=None,
                 expected_exception=None,
                 name=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._name = name
        self._state = STATE_CLOSED
        self._opened = datetime.utcnow()

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

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

        CircuitBreakerMonitor.register(self)

        @wraps(function)
        def wrapper(*args, **kwargs):
            return self.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
        """
        if self.opened:
            raise CircuitBreakerError(self)
        try:
            result = func(*args, **kwargs)
        except self._expected_exception as e:
            self._last_failure = e
            self.__call_failed()
            raise

        self.__call_succeeded()
        return result

    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 = datetime.utcnow()

    @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 datetime, when the circuit breaker will try to recover
        :return: datetime
        """
        return self._opened + timedelta(seconds=self._recovery_timeout)

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

    @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

    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,
            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)


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

fabfuel / circuitbreaker

Push — master ( 6a4cde...f28c44 )

circuitbreaker.CircuitBreaker.last_failure() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like