kytos.core.retry.for_all_methods() - Code Metrics - Inspection of "[Feature] kytos.core.retry decorators" - kytos-ng/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#216)

by Vinicius

created 2022-05-20 18:09 UTC

kytos.core.retry.for_all_methods() A

↳ Parent: kytos.core.retry

Complexity

Conditions

Size

Total Lines	15
Code Lines	11

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	6
CRAP Score	2

Importance

Changes

Metric	Value
cc	2
eloc	11
nop	2
dl	0
loc	15
ccs	6
cts	6
cp	1
crap	2
rs	9.85
c	0
b	0
f	0

"""Retry utilities module."""
import logging
from functools import wraps
from typing import Callable

from tenacity import retry

LOG = logging.getLogger(__name__)


def before_sleep(state) -> None:
    """Before sleep function for tenacity to also logs args and kwargs."""
    LOG.warning(
        f"Retry #{state.attempt_number} for {state.fn.__name__}, "
        f"args: {state.args}, kwargs: {state.kwargs}, "
        f"seconds since start: {state.seconds_since_start:.2f}",
    )


def retries(func: Callable, **tenacity_kwargs) -> Callable:
    """Retries decorator."""
    @retry(**tenacity_kwargs)
    @wraps(func)
    def decorated(*args, **kwargs):
        return func(*args, **kwargs)

    return decorated


def for_all_methods(decorator: Callable, **kwargs) -> Callable:
    """Decorator for all methods."""

    def decorated(cls):
        for attr in [
            name
            for name in dir(cls)
            if callable(getattr(cls, name))
            and not name.startswith("_")
            and not hasattr(getattr(cls, name), "__wrapped__")
        ]:
            setattr(cls, attr, decorator(getattr(cls, attr), **kwargs))
        return cls

    return decorated


1		"""Retry utilities module."""
2	1	import logging
3	1	from functools import wraps
4	1	from typing import Callable
5
6	1	from tenacity import retry
7
8	1	LOG = logging.getLogger(__name__)
9
10
11	1	def before_sleep(state) -> None:
12		"""Before sleep function for tenacity to also logs args and kwargs."""
13	1	LOG.warning(
14		f"Retry #{state.attempt_number} for {state.fn.__name__}, "
15		f"args: {state.args}, kwargs: {state.kwargs}, "
16		f"seconds since start: {state.seconds_since_start:.2f}",
17		)
18
19
20	1	def retries(func: Callable, **tenacity_kwargs) -> Callable:
21		"""Retries decorator."""
22	1	@retry(**tenacity_kwargs)
23	1	@wraps(func)
24	1	def decorated(args, *kwargs):
25	1	return func(args, *kwargs)
26
27	1	return decorated
28
29
30	1	def for_all_methods(decorator: Callable, **kwargs) -> Callable:
31		"""Decorator for all methods."""
32
33	1	def decorated(cls):
34	1	for attr in [
35		name
36		for name in dir(cls)
37		if callable(getattr(cls, name))
38		and not name.startswith("_")
39		and not hasattr(getattr(cls, name), "__wrapped__")
40		]:
41	1	setattr(cls, attr, decorator(getattr(cls, attr), **kwargs))
42	1	return cls
43
44		return decorated
45

kytos-ng / kytos

Pull Request — master (#216)

kytos.core.retry.for_all_methods() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like