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

Test Failed

Pull Request — master (#216)

by Vinicius

created 2022-05-20 18:03 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 %

Importance

Changes

Metric	Value
cc	2
eloc	11
nop	2
dl	0
loc	15
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			import logging
3			from functools import wraps
4			from typing import Callable
5
6			from tenacity import retry
7
8			LOG = logging.getLogger(__name__)
9
10
11			def before_sleep(state) -> None:
12			"""Before sleep function for tenacity to also logs args and kwargs."""
13			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			def retries(func: Callable, **tenacity_kwargs) -> Callable:
21			"""Retries decorator."""
22			@retry(**tenacity_kwargs)
23			@wraps(func)
24			def decorated(args, *kwargs):
25			return func(args, *kwargs)
26
27			return decorated
28
29
30			def for_all_methods(decorator: Callable, **kwargs) -> Callable:
31			"""Decorator for all methods."""
32
33			def decorated(cls):
34			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			setattr(cls, attr, decorator(getattr(cls, attr), **kwargs))
42			return cls
43
44			return decorated
45

kytos-ng / kytos

Pull Request — master (#216)

kytos.core.retry.for_all_methods() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like