blob: cb011b246fab1c65f41745304533f2c5d30b3892 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
|
import inspect
import typing as t
from functools import wraps
from .utils import _PassArg
from .utils import pass_eval_context
if t.TYPE_CHECKING:
V = t.TypeVar("V")
def async_variant(normal_func):
def decorator(async_func):
pass_arg = _PassArg.from_obj(normal_func)
need_eval_context = pass_arg is None
if pass_arg is _PassArg.environment:
def is_async(args):
return args[0].is_async
else:
def is_async(args):
return args[0].environment.is_async
@wraps(normal_func)
def wrapper(*args, **kwargs):
b = is_async(args)
if need_eval_context:
args = args[1:]
if b:
return async_func(*args, **kwargs)
return normal_func(*args, **kwargs)
if need_eval_context:
wrapper = pass_eval_context(wrapper)
wrapper.jinja_async_variant = True
return wrapper
return decorator
async def auto_await(value):
if inspect.isawaitable(value):
return await value
return value
async def auto_aiter(iterable):
if hasattr(iterable, "__aiter__"):
async for item in iterable:
yield item
else:
for item in iterable:
yield item
async def auto_to_list(
value: "t.Union[t.AsyncIterable[V], t.Iterable[V]]",
) -> "t.List[V]":
seq = []
if hasattr(value, "__aiter__"):
async for item in t.cast(t.AsyncIterable, value):
seq.append(item)
else:
for item in t.cast(t.Iterable, value):
seq.append(item)
return seq
|
