Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/jinja2/async_utils.py: 48%

1import inspect 

2import typing as t 

3from functools import WRAPPER_ASSIGNMENTS 

4from functools import wraps 

5 

6from .utils import _PassArg 

7from .utils import pass_eval_context 

8 

9V = t.TypeVar("V") 

10 

11 

12def async_variant(normal_func): # type: ignore 

13 def decorator(async_func): # type: ignore 

14 pass_arg = _PassArg.from_obj(normal_func) 

15 need_eval_context = pass_arg is None 

16 

17 if pass_arg is _PassArg.environment: 

18 

19 def is_async(args: t.Any) -> bool: 

20 return t.cast(bool, args[0].is_async) 

21 

22 else: 

23 

24 def is_async(args: t.Any) -> bool: 

25 return t.cast(bool, args[0].environment.is_async) 

26 

27 # Take the doc and annotations from the sync function, but the 

28 # name from the async function. Pallets-Sphinx-Themes 

29 # build_function_directive expects __wrapped__ to point to the 

30 # sync function. 

31 async_func_attrs = ("__module__", "__name__", "__qualname__") 

32 normal_func_attrs = tuple(set(WRAPPER_ASSIGNMENTS).difference(async_func_attrs)) 

33 

34 @wraps(normal_func, assigned=normal_func_attrs) 

35 @wraps(async_func, assigned=async_func_attrs, updated=()) 

36 def wrapper(*args, **kwargs): # type: ignore 

37 b = is_async(args) 

38 

39 if need_eval_context: 

40 args = args[1:] 

41 

42 if b: 

43 return async_func(*args, **kwargs) 

44 

45 return normal_func(*args, **kwargs) 

46 

47 if need_eval_context: 

48 wrapper = pass_eval_context(wrapper) 

49 

50 wrapper.jinja_async_variant = True 

51 return wrapper 

52 

53 return decorator 

54 

55 

56_common_primitives = {int, float, bool, str, list, dict, tuple, type(None)} 

57 

58 

59async def auto_await(value: t.Union[t.Awaitable["V"], "V"]) -> "V": 

60 # Avoid a costly call to isawaitable 

61 if type(value) in _common_primitives: 

62 return t.cast("V", value) 

63 

64 if inspect.isawaitable(value): 

65 return await t.cast("t.Awaitable[V]", value) 

66 

67 return t.cast("V", value) 

68 

69 

70async def auto_aiter( 

71 iterable: "t.Union[t.AsyncIterable[V], t.Iterable[V]]", 

72) -> "t.AsyncIterator[V]": 

73 if hasattr(iterable, "__aiter__"): 

74 async for item in t.cast("t.AsyncIterable[V]", iterable): 

75 yield item 

76 else: 

77 for item in t.cast("t.Iterable[V]", iterable): 

78 yield item 

79 

80 

81async def auto_to_list( 

82 value: "t.Union[t.AsyncIterable[V], t.Iterable[V]]", 

83) -> t.List["V"]: 

84 return [x async for x in auto_aiter(value)]