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

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