geronimo-iia /
async-rx
| Conditions | 7 |
| Total Lines | 56 |
| Code Lines | 24 |
| Lines | 0 |
| Ratio | 0 % |
| Changes | 0 | ||
Small methods make your code easier to understand, in particular if combined with a good name. Besides, if your method is small, finding a good name is usually much easier.
For example, if you find yourself adding comments to a method's body, this is usually a good sign to extract the commented part to a new method, and use the comment as a starting point when coming up with a good name for this new method.
Commonly applied refactorings include:
If many parameters/temporary variables are present:
| 1 | from typing import Any, Dict |
||
| 10 | def rx_from(observable_input: Any) -> Observable: |
||
| 11 | """Convert almost anything to an Observable. |
||
| 12 | |||
| 13 | Anything means: |
||
| 14 | - a dictionnary in an rx_dict |
||
| 15 | - an async iterable |
||
| 16 | - an iterable |
||
| 17 | - something which can be cast to an Observable (have a subscribe function) |
||
| 18 | - an object |
||
| 19 | |||
| 20 | Args: |
||
| 21 | observable_input (Any): A subscribable object |
||
| 22 | |||
| 23 | Returns: |
||
| 24 | (Observable): The Observable whose values are originally from the input object |
||
| 25 | that was converted. |
||
| 26 | |||
| 27 | """ |
||
| 28 | if isinstance(observable_input, Dict): |
||
| 29 | return rx_dict(initial_value=observable_input) |
||
| 30 | |||
| 31 | if hasattr(observable_input, "subscribe"): |
||
| 32 | # observable like |
||
| 33 | return rx_create(subscribe=observable_input.subscribe) |
||
| 34 | |||
| 35 | if hasattr(observable_input, "__aiter__"): |
||
| 36 | # something which be async iterable |
||
| 37 | async def _subscribe_aiter(an_observer: Observer) -> Subscription: |
||
| 38 | async for item in observable_input: |
||
| 39 | await an_observer.on_next(item) |
||
| 40 | await an_observer.on_completed() |
||
| 41 | |||
| 42 | return default_subscription |
||
| 43 | |||
| 44 | return rx_create(subscribe=_subscribe_aiter) |
||
| 45 | |||
| 46 | if hasattr(observable_input, "__iter__"): |
||
| 47 | # something iterable |
||
| 48 | async def _subscribe_iter(an_observer: Observer) -> Subscription: |
||
| 49 | for item in observable_input: |
||
| 50 | await an_observer.on_next(item) |
||
| 51 | await an_observer.on_completed() |
||
| 52 | |||
| 53 | return default_subscription |
||
| 54 | |||
| 55 | return rx_create(subscribe=_subscribe_iter) |
||
| 56 | |||
| 57 | # Build an simple singleton |
||
| 58 | |||
| 59 | async def _subscribe_object(an_observer: Observer) -> Subscription: |
||
| 60 | await an_observer.on_next(observable_input) |
||
| 61 | await an_observer.on_completed() |
||
| 62 | |||
| 63 | return default_subscription |
||
| 64 | |||
| 65 | return rx_create(subscribe=_subscribe_object) |
||
| 66 |
