jinja2.async_utils 

 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  # type: ignore[attr-defined]
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 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)]
def async_variant(normal_func): 
13def async_variant(normal_func):  # type: ignore
14    def decorator(async_func):  # type: ignore
15        pass_arg = _PassArg.from_obj(normal_func)
16        need_eval_context = pass_arg is None
17
18        if pass_arg is _PassArg.environment:
19
20            def is_async(args: t.Any) -> bool:
21                return t.cast(bool, args[0].is_async)
22
23        else:
24
25            def is_async(args: t.Any) -> bool:
26                return t.cast(bool, args[0].environment.is_async)
27
28        # Take the doc and annotations from the sync function, but the
29        # name from the async function. Pallets-Sphinx-Themes
30        # build_function_directive expects __wrapped__ to point to the
31        # sync function.
32        async_func_attrs = ("__module__", "__name__", "__qualname__")
33        normal_func_attrs = tuple(set(WRAPPER_ASSIGNMENTS).difference(async_func_attrs))
34
35        @wraps(normal_func, assigned=normal_func_attrs)
36        @wraps(async_func, assigned=async_func_attrs, updated=())
37        def wrapper(*args, **kwargs):  # type: ignore
38            b = is_async(args)
39
40            if need_eval_context:
41                args = args[1:]
42
43            if b:
44                return async_func(*args, **kwargs)
45
46            return normal_func(*args, **kwargs)
47
48        if need_eval_context:
49            wrapper = pass_eval_context(wrapper)
50
51        wrapper.jinja_async_variant = True  # type: ignore[attr-defined]
52        return wrapper
53
54    return decorator
async def auto_await(value: Union[Awaitable[~V], ~V]) -> ~V: 
60async def auto_await(value: t.Union[t.Awaitable["V"], "V"]) -> "V":
61    # Avoid a costly call to isawaitable
62    if type(value) in _common_primitives:
63        return t.cast("V", value)
64
65    if inspect.isawaitable(value):
66        return await t.cast("t.Awaitable[V]", value)
67
68    return t.cast("V", value)
async def auto_aiter(iterable: Union[AsyncIterable[~V], Iterable[~V]]) -> AsyncIterator[~V]: 
71async def auto_aiter(
72    iterable: "t.Union[t.AsyncIterable[V], t.Iterable[V]]",
73) -> "t.AsyncIterator[V]":
74    if hasattr(iterable, "__aiter__"):
75        async for item in t.cast("t.AsyncIterable[V]", iterable):
76            yield item
77    else:
78        for item in iterable:
79            yield item
async def auto_to_list(value: Union[AsyncIterable[~V], Iterable[~V]]) -> List[~V]: 
82async def auto_to_list(
83    value: "t.Union[t.AsyncIterable[V], t.Iterable[V]]",
84) -> t.List["V"]:
85    return [x async for x in auto_aiter(value)]