Total Complexity | 3 |
Total Lines | 59 |
Duplicated Lines | 0 % |
Changes | 0 |
1 | """Curiosity module define special construct with curio framework.""" |
||
2 | |||
3 | from typing import List |
||
4 | |||
5 | from .definition import ( |
||
6 | FAILURE, |
||
7 | SUCCESS, |
||
8 | AsyncInnerFunction, |
||
9 | CallableFunction, |
||
10 | node_metadata, |
||
11 | ) |
||
12 | from .utils import amap |
||
13 | |||
14 | |||
15 | __all__ = ['parallele'] |
||
16 | |||
17 | |||
18 | try: |
||
19 | from curio import gather, spawn |
||
20 | |||
21 | def parallele( |
||
22 | children: List[CallableFunction], succes_threshold: int = -1 |
||
23 | ) -> AsyncInnerFunction: |
||
24 | """ |
||
25 | Return an awaitable function which run children in parallele. |
||
26 | |||
27 | succes_threshold generalize traditional sequence/fallback. |
||
28 | succes_threshold must be in [0, len(children)], |
||
29 | default value is len(children) |
||
30 | |||
31 | if #success = succes_threshold, return a success |
||
32 | |||
33 | if #failure = len(children) - succes_threshold, return a failure |
||
34 | |||
35 | :param children: list of Awaitable |
||
36 | :param succes_threshold: succes threshold value |
||
37 | :return: an awaitable function. |
||
38 | |||
39 | """ |
||
40 | succes_threshold = succes_threshold if succes_threshold else len(children) |
||
41 | assert 0 <= succes_threshold <= len(children) |
||
42 | |||
43 | @node_metadata(properties=['succes_threshold']) |
||
44 | async def _parallele(): |
||
45 | |||
46 | results = await gather([task async for task in amap(spawn, children)]) |
||
47 | |||
48 | success = len(list(filter(bool, results))) |
||
49 | |||
50 | return SUCCESS if success >= succes_threshold else FAILURE |
||
51 | |||
52 | return _parallele |
||
53 | |||
54 | |||
55 | except Exception: # pylint: disable=broad-except |
||
56 | |||
57 | # default to a simple sequence |
||
58 | from .control import sequence as parallele |
||
59 |