ACIL FM

Nama	Ukuran	Aksi
__pycache__	-	Open
async_utils.py	2.41 MB	View DL Edit
bccache.py	13.73 MB	View DL Edit
compiler.py	70.48 MB	View DL Edit
constants.py	1.4 MB	View DL Edit
debug.py	6.15 MB	View DL Edit
defaults.py	1.24 MB	View DL Edit
environment.py	59.91 MB	View DL Edit
exceptions.py	4.95 MB	View DL Edit
ext.py	30.76 MB	View DL Edit
filters.py	52.25 MB	View DL Edit
idtracking.py	10.45 MB	View DL Edit
lexer.py	29.03 MB	View DL Edit
loaders.py	22.66 MB	View DL Edit
meta.py	4.29 MB	View DL Edit
nativetypes.py	4.13 MB	View DL Edit
nodes.py	33.74 MB	View DL Edit
optimizer.py	1.61 MB	View DL Edit
parser.py	38.67 MB	View DL Edit
py.typed	0 B	View DL Edit
runtime.py	32.69 MB	View DL Edit
sandbox.py	14.24 MB	View DL Edit
tests.py	5.77 MB	View DL Edit
utils.py	23.4 MB	View DL Edit
visitor.py	3.48 MB	View DL Edit
_identifier.py	1.91 MB	View DL Edit
__init__.py	1.88 MB	View DL Edit

Edit file: /usr/lib/python3.9/site-packages/ansible/_vendor/jinja2/async_utils.py

import inspect
import typing as t
from functools import WRAPPER_ASSIGNMENTS
from functools import wraps

from .utils import _PassArg
from .utils import pass_eval_context

V = t.TypeVar("V")

def async_variant(normal_func):  # type: ignore
    def decorator(async_func):  # type: ignore
        pass_arg = _PassArg.from_obj(normal_func)
        need_eval_context = pass_arg is None

if pass_arg is _PassArg.environment:

def is_async(args: t.Any) -> bool:
                return t.cast(bool, args[0].is_async)

else:

def is_async(args: t.Any) -> bool:
                return t.cast(bool, args[0].environment.is_async)

# Take the doc and annotations from the sync function, but the
        # name from the async function. Pallets-Sphinx-Themes
        # build_function_directive expects __wrapped__ to point to the
        # sync function.
        async_func_attrs = ("__module__", "__name__", "__qualname__")
        normal_func_attrs = tuple(set(WRAPPER_ASSIGNMENTS).difference(async_func_attrs))

@wraps(normal_func, assigned=normal_func_attrs)
        @wraps(async_func, assigned=async_func_attrs, updated=())
        def wrapper(*args, **kwargs):  # type: ignore
            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

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

async def auto_await(value: t.Union[t.Awaitable["V"], "V"]) -> "V":
    # Avoid a costly call to isawaitable
    if type(value) in _common_primitives:
        return t.cast("V", value)

if inspect.isawaitable(value):
        return await t.cast("t.Awaitable[V]", value)

return t.cast("V", value)

async def auto_aiter(
    iterable: "t.Union[t.AsyncIterable[V], t.Iterable[V]]",
) -> "t.AsyncIterator[V]":
    if hasattr(iterable, "__aiter__"):
        async for item in t.cast("t.AsyncIterable[V]", iterable):
            yield item
    else:
        for item in t.cast("t.Iterable[V]", iterable):
            yield item

async def auto_to_list(
    value: "t.Union[t.AsyncIterable[V], t.Iterable[V]]",
) -> t.List["V"]:
    return [x async for x in auto_aiter(value)]

Batal