venv/lib/python3.11/site-packages/jinja2/async_utils.py

   1 import inspect
   2 import typing as t
   3 from functools import WRAPPER_ASSIGNMENTS
   4 from functools import wraps
   5
   6 from .utils import _PassArg
   7 from .utils import pass_eval_context
   8
   9 V = t.TypeVar("V")
  10
  11
  12 def 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
  59 async 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
  70 async 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
  81 async 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)]