async_btree.decorator.retry() - Code Metrics - Inspection of "Documentation update" - geronimo-iia/async-btree - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#9)

by Guibert

created 2019-08-29 14:51 UTC

async_btree.decorator.retry() C

↳ Parent: async_btree.decorator

Complexity

Conditions

Size

Total Lines	40
Code Lines	21

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	9
eloc	21
nop	2
dl	0
loc	40
rs	6.6666
c	0
b	0
f	0

"""Decorator module define all decorator function node."""
from typing import Any

from .definition import (
    FAILURE,
    SUCCESS,
    AsyncInnerFunction,
    CallableFunction,
    ExceptionDecorator,
    node_metadata,
)


__all__ = [
    'alias',
    'decorate',
    'always_success',
    'always_failure',
    'is_success',
    'is_failure',
    'inverter',
    'retry',
    'retry_until_success',
    'retry_until_failed',
]


def alias(child: CallableFunction, name: str) -> AsyncInnerFunction:
    """Define an alias on our child.

    # Parameters
    child (CallableFunction): child function to decorate
    name (str): name of function tree

    # Returns
    (AsyncInnerFunction): an awaitable function.
    """

    @node_metadata(name=name)
    async def _alias():
        return await child()

    return _alias


def decorate(
    child: CallableFunction, decorator: CallableFunction, **kwargs
) -> AsyncInnerFunction:
    """Create a decorator.

    Post process a child with specified decorator function.
    First argument of decorator function must be a child.

    This method implement a simple lazy evaluation.

    # Parameters
    child (CallableFunction): child function to decorate
    decorator (CallableFunction): awaitable target decorator
    kwargs: optional keyed argument to pass to decorator function

    # Returns
    (AsyncInnerFunction): an awaitable function which
        return decorator evaluation against child.
    """

    @node_metadata(properties=['decorator'])
    async def _decorate():
        return await decorator(await child(), **kwargs)

    return _decorate


def always_success(child: CallableFunction) -> AsyncInnerFunction:
    """Create a node which always return SUCCESS value.

    # Parameters
    child (CallableFunction): child function to decorate

    # Returns
    (AsyncInnerFunction): an awaitable function which return child result if is truthy
        else SUCCESS (Any exception will be ignored).

    """

    @node_metadata()
    async def _always_success():
        result: Any = SUCCESS

        try:
            child_result = await child()
            if child_result:
                result = child_result

        except Exception:
            return result

        return result

    return _always_success


def always_failure(child: CallableFunction) -> AsyncInnerFunction:  # -> Awaitable:
    """Produce a function which always return FAILURE value.

    # Parameters
    child (CallableFunction): child function to decorate

    # Returns
    (AsyncInnerFunction): an awaitable function which return child result if is falsy
        else FAILURE, or a ControlFlowException if error occurs.

    """

    @node_metadata()
    async def _always_failure():
        result: Any = FAILURE

        try:
            child_result = await child()
            if not child_result:
                result = child_result

        except Exception as e:
            result = ExceptionDecorator(e)

        return result

    return _always_failure


def is_success(child: CallableFunction) -> AsyncInnerFunction:
    """Create a conditional node which test if child success.

    # Parameters
    child (CallableFunction): child function to decorate

    # Returns
    (AsyncInnerFunction): an awaitable function which return SUCCESS if child
        return SUCCESS else FAILURE.
        An exception will be evaluated as falsy.
    """

    @node_metadata()
    async def _is_success():
        try:
            return SUCCESS if bool(await child()) else FAILURE
        except Exception as e:
            return ExceptionDecorator(e)

    return _is_success


def is_failure(child: CallableFunction) -> AsyncInnerFunction:
    """Create a conditional node which test if child fail.

    # Parameters
    child (CallableFunction): child function to decorate

    # Returns
    (AsyncInnerFunction): an awaitable function which return SUCCESS if child
        return FAILURE else FAILURE.
        An exception will be evaluated as a success.
    """

    @node_metadata()
    async def _is_failure():
        try:
            return SUCCESS if not bool(await child()) else FAILURE
        except Exception:  # pylint: disable=broad-except
            return SUCCESS

    return _is_failure


def inverter(child: CallableFunction) -> AsyncInnerFunction:
    """Invert node status.

    # Parameters
    child (CallableFunction): child function to decorate

    # Returns
    (AsyncInnerFunction): an awaitable function which return SUCCESS if child
        return FAILURE else SUCCESS
    """

    @node_metadata()
    async def _inverter():
        return not await child()

    return _inverter


def retry(child: CallableFunction, max_retry: int = 3) -> AsyncInnerFunction:
    """Retry child evaluation at most max_retry time on failure until child succeed.

    # Parameters
    child (CallableFunction): child function to decorate
    max_retry (int): max retry count (default 3), -1 mean infinite retry

    # Returns
    (AsyncInnerFunction): an awaitable function which retry child evaluation
        at most max_retry time on failure until child succeed.
        If max_retry is reached, returns FAILURE or last exception.
    """
    if not (max_retry > 0 or max_retry == -1):
        raise AssertionError('max_retry')

    @node_metadata(properties=['max_retry'])
    async def _retry():
        retry_count = 0
        infinite_retry_condition = max_retry == -1
        result: Any = FAILURE

        while infinite_retry_condition or retry_count < max_retry:
            try:
                result = await child()
                if result:
                    return result

            except Exception as e:
                # return last failure exception
                if (
                    not infinite_retry_condition
                ):  # avoid data allocation if never returned
                    result = ExceptionDecorator(e)

            if not infinite_retry_condition:  # avoid overflow
                retry_count += 1

        return result

    return _retry


def retry_until_success(child: CallableFunction) -> AsyncInnerFunction:
    """Retry child until success.

    # Parameters
    child (CallableFunction): child function to decorate

    # Returns
    (AsyncInnerFunction): an awaitable function which try to evaluate child
        until it succeed.
    """

    # @node_metadata()
    # async def _retry_until_success():
    #    return await retry(child=child, max_retry=-1)()

    return node_metadata(name='retry_until_success')(retry(child=child, max_retry=-1))


def retry_until_failed(child: CallableFunction) -> AsyncInnerFunction:
    """Retry child until failed.

    # Parameters
    child (CallableFunction): child function to decorate

    # Returns
    (AsyncInnerFunction): an awaitable function which try to evaluate child
        until it failed.
    """

    return node_metadata(name='retry_until_failed')(
        retry(child=inverter(child), max_retry=-1)
    )


1			"""Decorator module define all decorator function node."""
2			from typing import Any
3
4			from .definition import (
5			FAILURE,
6			SUCCESS,
7			AsyncInnerFunction,
8			CallableFunction,
9			ExceptionDecorator,
10			node_metadata,
11			)
12
13
14			__all__ = [
15			'alias',
16			'decorate',
17			'always_success',
18			'always_failure',
19			'is_success',
20			'is_failure',
21			'inverter',
22			'retry',
23			'retry_until_success',
24			'retry_until_failed',
25			]
26
27
28			def alias(child: CallableFunction, name: str) -> AsyncInnerFunction:
29			"""Define an alias on our child.
30
31			# Parameters
32			child (CallableFunction): child function to decorate
33			name (str): name of function tree
34
35			# Returns
36			(AsyncInnerFunction): an awaitable function.
37			"""
38
39			@node_metadata(name=name)
40			async def _alias():
41			return await child()
42
43			return _alias
44
45
46			def decorate(
47			child: CallableFunction, decorator: CallableFunction, **kwargs
48			) -> AsyncInnerFunction:
49			"""Create a decorator.
50
51			Post process a child with specified decorator function.
52			First argument of decorator function must be a child.
53
54			This method implement a simple lazy evaluation.
55
56			# Parameters
57			child (CallableFunction): child function to decorate
58			decorator (CallableFunction): awaitable target decorator
59			kwargs: optional keyed argument to pass to decorator function
60
61			# Returns
62			(AsyncInnerFunction): an awaitable function which
63			return decorator evaluation against child.
64			"""
65
66			@node_metadata(properties=['decorator'])
67			async def _decorate():
68			return await decorator(await child(), **kwargs)
69
70			return _decorate
71
72
73			def always_success(child: CallableFunction) -> AsyncInnerFunction:
74			"""Create a node which always return SUCCESS value.
75
76			# Parameters
77			child (CallableFunction): child function to decorate
78
79			# Returns
80			(AsyncInnerFunction): an awaitable function which return child result if is truthy
81			else SUCCESS (Any exception will be ignored).
82
83			"""
84
85			@node_metadata()
86			async def _always_success():
87			result: Any = SUCCESS
88
89			try:
90			child_result = await child()
91			if child_result:
92			result = child_result
93
94			except Exception:
95			return result
96
97			return result
98
99			return _always_success
100
101
102			def always_failure(child: CallableFunction) -> AsyncInnerFunction: # -> Awaitable:
103			"""Produce a function which always return FAILURE value.
104
105			# Parameters
106			child (CallableFunction): child function to decorate
107
108			# Returns
109			(AsyncInnerFunction): an awaitable function which return child result if is falsy
110			else FAILURE, or a ControlFlowException if error occurs.
111
112			"""
113
114			@node_metadata()
115			async def _always_failure():
116			result: Any = FAILURE
117
118			try:
119			child_result = await child()
120			if not child_result:
121			result = child_result
122
123			except Exception as e:
124			result = ExceptionDecorator(e)
125
126			return result
127
128			return _always_failure
129
130
131			def is_success(child: CallableFunction) -> AsyncInnerFunction:
132			"""Create a conditional node which test if child success.
133
134			# Parameters
135			child (CallableFunction): child function to decorate
136
137			# Returns
138			(AsyncInnerFunction): an awaitable function which return SUCCESS if child
139			return SUCCESS else FAILURE.
140			An exception will be evaluated as falsy.
141			"""
142
143			@node_metadata()
144			async def _is_success():
145			try:
146			return SUCCESS if bool(await child()) else FAILURE
147			except Exception as e:
148			return ExceptionDecorator(e)
149
150			return _is_success
151
152
153			def is_failure(child: CallableFunction) -> AsyncInnerFunction:
154			"""Create a conditional node which test if child fail.
155
156			# Parameters
157			child (CallableFunction): child function to decorate
158
159			# Returns
160			(AsyncInnerFunction): an awaitable function which return SUCCESS if child
161			return FAILURE else FAILURE.
162			An exception will be evaluated as a success.
163			"""
164
165			@node_metadata()
166			async def _is_failure():
167			try:
168			return SUCCESS if not bool(await child()) else FAILURE
169			except Exception: # pylint: disable=broad-except
170			return SUCCESS
171
172			return _is_failure
173
174
175			def inverter(child: CallableFunction) -> AsyncInnerFunction:
176			"""Invert node status.
177
178			# Parameters
179			child (CallableFunction): child function to decorate
180
181			# Returns
182			(AsyncInnerFunction): an awaitable function which return SUCCESS if child
183			return FAILURE else SUCCESS
184			"""
185
186			@node_metadata()
187			async def _inverter():
188			return not await child()
189
190			return _inverter
191
192
193			def retry(child: CallableFunction, max_retry: int = 3) -> AsyncInnerFunction:
194			"""Retry child evaluation at most max_retry time on failure until child succeed.
195
196			# Parameters
197			child (CallableFunction): child function to decorate
198			max_retry (int): max retry count (default 3), -1 mean infinite retry
199
200			# Returns
201			(AsyncInnerFunction): an awaitable function which retry child evaluation
202			at most max_retry time on failure until child succeed.
203			If max_retry is reached, returns FAILURE or last exception.
204			"""
205			if not (max_retry > 0 or max_retry == -1):
206			raise AssertionError('max_retry')
207
208			@node_metadata(properties=['max_retry'])
209			async def _retry():
210			retry_count = 0
211			infinite_retry_condition = max_retry == -1
212			result: Any = FAILURE
213
214			while infinite_retry_condition or retry_count < max_retry:
215			try:
216			result = await child()
217			if result:
218			return result
219
220			except Exception as e:
221			# return last failure exception
222			if (
223			not infinite_retry_condition
224			): # avoid data allocation if never returned
225			result = ExceptionDecorator(e)
226
227			if not infinite_retry_condition: # avoid overflow
228			retry_count += 1
229
230			return result
231
232			return _retry
233
234
235			def retry_until_success(child: CallableFunction) -> AsyncInnerFunction:
236			"""Retry child until success.
237
238			# Parameters
239			child (CallableFunction): child function to decorate
240
241			# Returns
242			(AsyncInnerFunction): an awaitable function which try to evaluate child
243			until it succeed.
244			"""
245
246			# @node_metadata()
247			# async def _retry_until_success():
248			# return await retry(child=child, max_retry=-1)()
249
250			return node_metadata(name='retry_until_success')(retry(child=child, max_retry=-1))
251
252
253			def retry_until_failed(child: CallableFunction) -> AsyncInnerFunction:
254			"""Retry child until failed.
255
256			# Parameters
257			child (CallableFunction): child function to decorate
258
259			# Returns
260			(AsyncInnerFunction): an awaitable function which try to evaluate child
261			until it failed.
262			"""
263
264			return node_metadata(name='retry_until_failed')(
265			retry(child=inverter(child), max_retry=-1)
266			)
267

geronimo-iia / async-btree

Pull Request — master (#9)

async_btree.decorator.retry() C

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like