async_btree.decorator.is_failure() - Code Metrics - Inspection of "Merge pull request #20 from geronimo-iia/v1.1.0" - geronimo-iia/async-btree - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( a1fcd1...4af13f )

by Guibert

created 2020-11-20 12:15 UTC

async_btree.decorator.is_failure() A

↳ Parent: async_btree.decorator

Complexity

Conditions

Size

Total Lines	18
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	6
nop	1
dl	0
loc	18
rs	10
c	0
b	0
f	0

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

from .definition import FAILURE, SUCCESS, AsyncInnerFunction, CallableFunction, ControlFlowException, node_metadata
from .utils import to_async

__all__ = [
    'alias',
    'decorate',
    'ignore_exception',
    '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.

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

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

    _child = to_async(child)

    # we use a dedicted function to 'duplicate' the child reference
    @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.

    Args:
        child (CallableFunction): child function to decorate
        decorator (CallableFunction): awaitable target decorator with profile 'decorator(child_result, **kwargs)'
        kwargs: optional keyed argument to pass to decorator function

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

    _child = to_async(child)
    _decorator = to_async(decorator)

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

    return _decorate


def ignore_exception(child: CallableFunction) -> AsyncInnerFunction:
    """Create a node which ignore runtime exception.

    Args:
        child (CallableFunction): child function to decorate

    Returns:
        (AsyncInnerFunction): an awaitable function which return child result
        or any exception with a falsy meaning in a ControlFlowException.

    """

    _child = to_async(child)

    @node_metadata()
    async def _ignore_exception():

        try:
            return await _child()

        except Exception as e:
            return ControlFlowException.instanciate(e)

    return _ignore_exception


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

    Args:
        child (CallableFunction): child function to decorate
        silent_exception (bool): if true then exception will be ignored

    Returns:
        (AsyncInnerFunction): an awaitable function which return child result if it is truthy
            else SUCCESS.

    Raises:
        ControlFlowException : if error occurs

    """

    _child = to_async(child)

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

        try:
            child_result = await _child()
            if bool(child_result):
                result = child_result

        except Exception as e:
            raise ControlFlowException.instanciate(e)

        return result

    return _always_success


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

    Args:
        child (CallableFunction): child function to decorate

    Returns:
        (AsyncInnerFunction): an awaitable function which return child result if is falsy
            else FAILURE.

    Raises:
        ControlFlowException : if error occurs

    """

    _child = to_async(child)

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

        try:
            child_result = await _child()
            if not bool(child_result):
                result = child_result

        except Exception as e:
            raise ControlFlowException.instanciate(e)

        return result

    return _always_failure


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

    Args:
        child (CallableFunction): child function to decorate

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

    _child = to_async(child)

    @node_metadata()
    async def _is_success():
        return SUCCESS if bool(await _child()) else FAILURE

    return _is_success


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

    Args:
        child (CallableFunction): child function to decorate

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

    _child = to_async(child)

    @node_metadata()
    async def _is_failure():
        return SUCCESS if not bool(await _child()) else FAILURE

    return _is_failure


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

    Args:
        child (CallableFunction): child function to decorate

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

    _child = to_async(child)

    @node_metadata()
    async def _inverter():
        return not bool(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.

    Args:
        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')

    _child = to_async(child)

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

        while not bool(result) and retry_count != 0:
            result = await _child()
            print(f"result : {result}")
            retry_count -= 1

        return result

    return _retry


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

    Args:
        child (CallableFunction): child function to decorate

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

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


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

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

geronimo-iia / async-btree

Push — master ( a1fcd1...4af13f )

async_btree.decorator.is_failure() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like