dev4py.utils.awaitables

 1"""
 2The `awaitables` module provides a set of utility functions to simplify Awaitable operations
 3"""
 4
 5# Copyright 2022 the original author or authors (i.e.: St4rG00se for Dev4py).
 6#
 7# Licensed under the Apache License, Version 2.0 (the "License");
 8# you may not use this file except in compliance with the License.
 9# You may obtain a copy of the License at
10#
11#      https://www.apache.org/licenses/LICENSE-2.0
12#
13# Unless required by applicable law or agreed to in writing, software
14# distributed under the License is distributed on an "AS IS" BASIS,
15# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16# See the License for the specific language governing permissions and
17# limitations under the License.
18
19
20from typing import Awaitable, Any, cast
21
22from dev4py.utils import objects
23from dev4py.utils.types import T, R, Function, SyncOrAsync
24
25
26def is_awaitable(value: Any) -> bool:
27    """
28    If the given value is an Awaitable (Coroutine, Task or Future), returns true, otherwise false
29
30    Returns:
31        bool: true if the given value is an Awaitable, otherwise false
32    """
33    return isinstance(value, Awaitable)
34
35
36def to_sync_or_async_param_function(
37        function: Function[T, SyncOrAsync[R]]
38) -> Function[SyncOrAsync[T], Awaitable[R]]:
39    """
40    If the given Function uses sync parameter, create a new equivalent async function supporting sync or Awaitable
41    parameter
42
43    Returned Function specificities:
44        1. Is an async function (it means if the given function result is R the new function result is an Awaitable[R])
45        2. If the given function is async (i.e. result is an Awaitable[R]), the new function result stay an Awaitable[R]
46        3. !WARNING! If the given function expected parameter is an Awaitable, it will be awaited by the wrapper before
47        being passed as given function parameter. So, it will probably raise an error depending on your code
48
49    Note: This function can be useful associated with the map function of a `JOptional` with Awaitable value
50
51    Args:
52        function: The function with sync parameter
53
54    Returns:
55        sync_or_async_param_function: An async function with awaitable parameter
56
57    Raises:
58        TypeError: Raises a TypeError if function is None
59        Exception: If the given function expected parameter is an Awaitable, it will be awaited by the wrapper before
60        being passed as function parameter, and, it will probably raise an error depending on function your code
61    """
62    objects.require_non_none(function)
63
64    async def sync_or_async_param_function(param: SyncOrAsync[T]) -> R:
65        result: SyncOrAsync[R] = function((await cast(Awaitable[T], param)) if is_awaitable(param) else cast(T, param))
66        return await cast(Awaitable[R], result) if is_awaitable(result) else cast(R, result)
67
68    return sync_or_async_param_function
69
70
71def to_awaitable(value: SyncOrAsync[T]) -> Awaitable[T]:
72    """
73    This function map a SyncOrAsync[T] to an Awaitable[T]
74
75    Note: This can be useful to work with SyncOrAsync values in order to always works with `await`. Without this
76    function, you always have to call is_awaitable in order to know if you have to use `await` or not.
77
78    Args:
79        value: the value to map
80
81    Returns:
82        Awaitable[T]: a value to await
83    """
84
85    async def _sync_to_awaitable(param: T) -> T:
86        return param
87
88    return cast(Awaitable[T], value) if is_awaitable(value) else _sync_to_awaitable(cast(T, value))
89
90
91async def async_none() -> None:
92    """
93    This function provides an Awaitable[None]
94
95    Returns:
96        Awaitable[None]: An awaitable of None
97
98    """
99    return None