diff --git a/venv/Lib/site-packages/jinja2/py.typed b/venv/Lib/site-packages/jinja2/py.typed new file mode 100644 index 0000000..06d7405 Binary files /dev/null and b/venv/Lib/site-packages/jinja2/py.typed differ diff --git a/venv/Lib/site-packages/jinja2/runtime.py b/venv/Lib/site-packages/jinja2/runtime.py new file mode 100644 index 0000000..09119e2 --- /dev/null +++ b/venv/Lib/site-packages/jinja2/runtime.py @@ -0,0 +1,1062 @@ +"""The runtime functions and state used by compiled templates.""" + +import functools +import sys +import typing as t +from collections import abc +from itertools import chain + +from markupsafe import escape # noqa: F401 +from markupsafe import Markup +from markupsafe import soft_str + +from .async_utils import auto_aiter +from .async_utils import auto_await # noqa: F401 +from .exceptions import TemplateNotFound # noqa: F401 +from .exceptions import TemplateRuntimeError # noqa: F401 +from .exceptions import UndefinedError +from .nodes import EvalContext +from .utils import _PassArg +from .utils import concat +from .utils import internalcode +from .utils import missing +from .utils import Namespace # noqa: F401 +from .utils import object_type_repr +from .utils import pass_eval_context + +V = t.TypeVar("V") +F = t.TypeVar("F", bound=t.Callable[..., t.Any]) + +if t.TYPE_CHECKING: + import logging + + import typing_extensions as te + + from .environment import Environment + + class LoopRenderFunc(te.Protocol): + def __call__( + self, + reciter: t.Iterable[V], + loop_render_func: "LoopRenderFunc", + depth: int = 0, + ) -> str: ... + + +# these variables are exported to the template runtime +exported = [ + "LoopContext", + "TemplateReference", + "Macro", + "Markup", + "TemplateRuntimeError", + "missing", + "escape", + "markup_join", + "str_join", + "identity", + "TemplateNotFound", + "Namespace", + "Undefined", + "internalcode", +] +async_exported = [ + "AsyncLoopContext", + "auto_aiter", + "auto_await", +] + + +def identity(x: V) -> V: + """Returns its argument. Useful for certain things in the + environment. + """ + return x + + +def markup_join(seq: t.Iterable[t.Any]) -> str: + """Concatenation that escapes if necessary and converts to string.""" + buf = [] + iterator = map(soft_str, seq) + for arg in iterator: + buf.append(arg) + if hasattr(arg, "__html__"): + return Markup("").join(chain(buf, iterator)) + return concat(buf) + + +def str_join(seq: t.Iterable[t.Any]) -> str: + """Simple args to string conversion and concatenation.""" + return concat(map(str, seq)) + + +def new_context( + environment: "Environment", + template_name: t.Optional[str], + blocks: t.Dict[str, t.Callable[["Context"], t.Iterator[str]]], + vars: t.Optional[t.Dict[str, t.Any]] = None, + shared: bool = False, + globals: t.Optional[t.MutableMapping[str, t.Any]] = None, + locals: t.Optional[t.Mapping[str, t.Any]] = None, +) -> "Context": + """Internal helper for context creation.""" + if vars is None: + vars = {} + if shared: + parent = vars + else: + parent = dict(globals or (), **vars) + if locals: + # if the parent is shared a copy should be created because + # we don't want to modify the dict passed + if shared: + parent = dict(parent) + for key, value in locals.items(): + if value is not missing: + parent[key] = value + return environment.context_class( + environment, parent, template_name, blocks, globals=globals + ) + + +class TemplateReference: + """The `self` in templates.""" + + def __init__(self, context: "Context") -> None: + self.__context = context + + def __getitem__(self, name: str) -> t.Any: + blocks = self.__context.blocks[name] + return BlockReference(name, self.__context, blocks, 0) + + def __repr__(self) -> str: + return f"<{type(self).__name__} {self.__context.name!r}>" + + +def _dict_method_all(dict_method: F) -> F: + @functools.wraps(dict_method) + def f_all(self: "Context") -> t.Any: + return dict_method(self.get_all()) + + return t.cast(F, f_all) + + +@abc.Mapping.register +class Context: + """The template context holds the variables of a template. It stores the + values passed to the template and also the names the template exports. + Creating instances is neither supported nor useful as it's created + automatically at various stages of the template evaluation and should not + be created by hand. + + The context is immutable. Modifications on :attr:`parent` **must not** + happen and modifications on :attr:`vars` are allowed from generated + template code only. Template filters and global functions marked as + :func:`pass_context` get the active context passed as first argument + and are allowed to access the context read-only. + + The template context supports read only dict operations (`get`, + `keys`, `values`, `items`, `iterkeys`, `itervalues`, `iteritems`, + `__getitem__`, `__contains__`). Additionally there is a :meth:`resolve` + method that doesn't fail with a `KeyError` but returns an + :class:`Undefined` object for missing variables. + """ + + def __init__( + self, + environment: "Environment", + parent: t.Dict[str, t.Any], + name: t.Optional[str], + blocks: t.Dict[str, t.Callable[["Context"], t.Iterator[str]]], + globals: t.Optional[t.MutableMapping[str, t.Any]] = None, + ): + self.parent = parent + self.vars: t.Dict[str, t.Any] = {} + self.environment: Environment = environment + self.eval_ctx = EvalContext(self.environment, name) + self.exported_vars: t.Set[str] = set() + self.name = name + self.globals_keys = set() if globals is None else set(globals) + + # create the initial mapping of blocks. Whenever template inheritance + # takes place the runtime will update this mapping with the new blocks + # from the template. + self.blocks = {k: [v] for k, v in blocks.items()} + + def super( + self, name: str, current: t.Callable[["Context"], t.Iterator[str]] + ) -> t.Union["BlockReference", "Undefined"]: + """Render a parent block.""" + try: + blocks = self.blocks[name] + index = blocks.index(current) + 1 + blocks[index] + except LookupError: + return self.environment.undefined( + f"there is no parent block called {name!r}.", name="super" + ) + return BlockReference(name, self, blocks, index) + + def get(self, key: str, default: t.Any = None) -> t.Any: + """Look up a variable by name, or return a default if the key is + not found. + + :param key: The variable name to look up. + :param default: The value to return if the key is not found. + """ + try: + return self[key] + except KeyError: + return default + + def resolve(self, key: str) -> t.Union[t.Any, "Undefined"]: + """Look up a variable by name, or return an :class:`Undefined` + object if the key is not found. + + If you need to add custom behavior, override + :meth:`resolve_or_missing`, not this method. The various lookup + functions use that method, not this one. + + :param key: The variable name to look up. + """ + rv = self.resolve_or_missing(key) + + if rv is missing: + return self.environment.undefined(name=key) + + return rv + + def resolve_or_missing(self, key: str) -> t.Any: + """Look up a variable by name, or return a ``missing`` sentinel + if the key is not found. + + Override this method to add custom lookup behavior. + :meth:`resolve`, :meth:`get`, and :meth:`__getitem__` use this + method. Don't call this method directly. + + :param key: The variable name to look up. + """ + if key in self.vars: + return self.vars[key] + + if key in self.parent: + return self.parent[key] + + return missing + + def get_exported(self) -> t.Dict[str, t.Any]: + """Get a new dict with the exported variables.""" + return {k: self.vars[k] for k in self.exported_vars} + + def get_all(self) -> t.Dict[str, t.Any]: + """Return the complete context as dict including the exported + variables. For optimizations reasons this might not return an + actual copy so be careful with using it. + """ + if not self.vars: + return self.parent + if not self.parent: + return self.vars + return dict(self.parent, **self.vars) + + @internalcode + def call( + __self, + __obj: t.Callable[..., t.Any], + *args: t.Any, + **kwargs: t.Any, # noqa: B902 + ) -> t.Union[t.Any, "Undefined"]: + """Call the callable with the arguments and keyword arguments + provided but inject the active context or environment as first + argument if the callable has :func:`pass_context` or + :func:`pass_environment`. + """ + if __debug__: + __traceback_hide__ = True # noqa + + # Allow callable classes to take a context + if ( + hasattr(__obj, "__call__") # noqa: B004 + and _PassArg.from_obj(__obj.__call__) is not None + ): + __obj = __obj.__call__ + + pass_arg = _PassArg.from_obj(__obj) + + if pass_arg is _PassArg.context: + # the active context should have access to variables set in + # loops and blocks without mutating the context itself + if kwargs.get("_loop_vars"): + __self = __self.derived(kwargs["_loop_vars"]) + if kwargs.get("_block_vars"): + __self = __self.derived(kwargs["_block_vars"]) + args = (__self,) + args + elif pass_arg is _PassArg.eval_context: + args = (__self.eval_ctx,) + args + elif pass_arg is _PassArg.environment: + args = (__self.environment,) + args + + kwargs.pop("_block_vars", None) + kwargs.pop("_loop_vars", None) + + try: + return __obj(*args, **kwargs) + except StopIteration: + return __self.environment.undefined( + "value was undefined because a callable raised a" + " StopIteration exception" + ) + + def derived(self, locals: t.Optional[t.Dict[str, t.Any]] = None) -> "Context": + """Internal helper function to create a derived context. This is + used in situations where the system needs a new context in the same + template that is independent. + """ + context = new_context( + self.environment, self.name, {}, self.get_all(), True, None, locals + ) + context.eval_ctx = self.eval_ctx + context.blocks.update((k, list(v)) for k, v in self.blocks.items()) + return context + + keys = _dict_method_all(dict.keys) + values = _dict_method_all(dict.values) + items = _dict_method_all(dict.items) + + def __contains__(self, name: str) -> bool: + return name in self.vars or name in self.parent + + def __getitem__(self, key: str) -> t.Any: + """Look up a variable by name with ``[]`` syntax, or raise a + ``KeyError`` if the key is not found. + """ + item = self.resolve_or_missing(key) + + if item is missing: + raise KeyError(key) + + return item + + def __repr__(self) -> str: + return f"<{type(self).__name__} {self.get_all()!r} of {self.name!r}>" + + +class BlockReference: + """One block on a template reference.""" + + def __init__( + self, + name: str, + context: "Context", + stack: t.List[t.Callable[["Context"], t.Iterator[str]]], + depth: int, + ) -> None: + self.name = name + self._context = context + self._stack = stack + self._depth = depth + + @property + def super(self) -> t.Union["BlockReference", "Undefined"]: + """Super the block.""" + if self._depth + 1 >= len(self._stack): + return self._context.environment.undefined( + f"there is no parent block called {self.name!r}.", name="super" + ) + return BlockReference(self.name, self._context, self._stack, self._depth + 1) + + @internalcode + async def _async_call(self) -> str: + rv = self._context.environment.concat( # type: ignore + [x async for x in self._stack[self._depth](self._context)] # type: ignore + ) + + if self._context.eval_ctx.autoescape: + return Markup(rv) + + return rv + + @internalcode + def __call__(self) -> str: + if self._context.environment.is_async: + return self._async_call() # type: ignore + + rv = self._context.environment.concat( # type: ignore + self._stack[self._depth](self._context) + ) + + if self._context.eval_ctx.autoescape: + return Markup(rv) + + return rv + + +class LoopContext: + """A wrapper iterable for dynamic ``for`` loops, with information + about the loop and iteration. + """ + + #: Current iteration of the loop, starting at 0. + index0 = -1 + + _length: t.Optional[int] = None + _after: t.Any = missing + _current: t.Any = missing + _before: t.Any = missing + _last_changed_value: t.Any = missing + + def __init__( + self, + iterable: t.Iterable[V], + undefined: t.Type["Undefined"], + recurse: t.Optional["LoopRenderFunc"] = None, + depth0: int = 0, + ) -> None: + """ + :param iterable: Iterable to wrap. + :param undefined: :class:`Undefined` class to use for next and + previous items. + :param recurse: The function to render the loop body when the + loop is marked recursive. + :param depth0: Incremented when looping recursively. + """ + self._iterable = iterable + self._iterator = self._to_iterator(iterable) + self._undefined = undefined + self._recurse = recurse + #: How many levels deep a recursive loop currently is, starting at 0. + self.depth0 = depth0 + + @staticmethod + def _to_iterator(iterable: t.Iterable[V]) -> t.Iterator[V]: + return iter(iterable) + + @property + def length(self) -> int: + """Length of the iterable. + + If the iterable is a generator or otherwise does not have a + size, it is eagerly evaluated to get a size. + """ + if self._length is not None: + return self._length + + try: + self._length = len(self._iterable) # type: ignore + except TypeError: + iterable = list(self._iterator) + self._iterator = self._to_iterator(iterable) + self._length = len(iterable) + self.index + (self._after is not missing) + + return self._length + + def __len__(self) -> int: + return self.length + + @property + def depth(self) -> int: + """How many levels deep a recursive loop currently is, starting at 1.""" + return self.depth0 + 1 + + @property + def index(self) -> int: + """Current iteration of the loop, starting at 1.""" + return self.index0 + 1 + + @property + def revindex0(self) -> int: + """Number of iterations from the end of the loop, ending at 0. + + Requires calculating :attr:`length`. + """ + return self.length - self.index + + @property + def revindex(self) -> int: + """Number of iterations from the end of the loop, ending at 1. + + Requires calculating :attr:`length`. + """ + return self.length - self.index0 + + @property + def first(self) -> bool: + """Whether this is the first iteration of the loop.""" + return self.index0 == 0 + + def _peek_next(self) -> t.Any: + """Return the next element in the iterable, or :data:`missing` + if the iterable is exhausted. Only peeks one item ahead, caching + the result in :attr:`_last` for use in subsequent checks. The + cache is reset when :meth:`__next__` is called. + """ + if self._after is not missing: + return self._after + + self._after = next(self._iterator, missing) + return self._after + + @property + def last(self) -> bool: + """Whether this is the last iteration of the loop. + + Causes the iterable to advance early. See + :func:`itertools.groupby` for issues this can cause. + The :func:`groupby` filter avoids that issue. + """ + return self._peek_next() is missing + + @property + def previtem(self) -> t.Union[t.Any, "Undefined"]: + """The item in the previous iteration. Undefined during the + first iteration. + """ + if self.first: + return self._undefined("there is no previous item") + + return self._before + + @property + def nextitem(self) -> t.Union[t.Any, "Undefined"]: + """The item in the next iteration. Undefined during the last + iteration. + + Causes the iterable to advance early. See + :func:`itertools.groupby` for issues this can cause. + The :func:`jinja-filters.groupby` filter avoids that issue. + """ + rv = self._peek_next() + + if rv is missing: + return self._undefined("there is no next item") + + return rv + + def cycle(self, *args: V) -> V: + """Return a value from the given args, cycling through based on + the current :attr:`index0`. + + :param args: One or more values to cycle through. + """ + if not args: + raise TypeError("no items for cycling given") + + return args[self.index0 % len(args)] + + def changed(self, *value: t.Any) -> bool: + """Return ``True`` if previously called with a different value + (including when called for the first time). + + :param value: One or more values to compare to the last call. + """ + if self._last_changed_value != value: + self._last_changed_value = value + return True + + return False + + def __iter__(self) -> "LoopContext": + return self + + def __next__(self) -> t.Tuple[t.Any, "LoopContext"]: + if self._after is not missing: + rv = self._after + self._after = missing + else: + rv = next(self._iterator) + + self.index0 += 1 + self._before = self._current + self._current = rv + return rv, self + + @internalcode + def __call__(self, iterable: t.Iterable[V]) -> str: + """When iterating over nested data, render the body of the loop + recursively with the given inner iterable data. + + The loop must have the ``recursive`` marker for this to work. + """ + if self._recurse is None: + raise TypeError( + "The loop must have the 'recursive' marker to be called recursively." + ) + + return self._recurse(iterable, self._recurse, depth=self.depth) + + def __repr__(self) -> str: + return f"<{type(self).__name__} {self.index}/{self.length}>" + + +class AsyncLoopContext(LoopContext): + _iterator: t.AsyncIterator[t.Any] # type: ignore + + @staticmethod + def _to_iterator( # type: ignore + iterable: t.Union[t.Iterable[V], t.AsyncIterable[V]], + ) -> t.AsyncIterator[V]: + return auto_aiter(iterable) + + @property + async def length(self) -> int: # type: ignore + if self._length is not None: + return self._length + + try: + self._length = len(self._iterable) # type: ignore + except TypeError: + iterable = [x async for x in self._iterator] + self._iterator = self._to_iterator(iterable) + self._length = len(iterable) + self.index + (self._after is not missing) + + return self._length + + @property + async def revindex0(self) -> int: # type: ignore + return await self.length - self.index + + @property + async def revindex(self) -> int: # type: ignore + return await self.length - self.index0 + + async def _peek_next(self) -> t.Any: + if self._after is not missing: + return self._after + + try: + self._after = await self._iterator.__anext__() + except StopAsyncIteration: + self._after = missing + + return self._after + + @property + async def last(self) -> bool: # type: ignore + return await self._peek_next() is missing + + @property + async def nextitem(self) -> t.Union[t.Any, "Undefined"]: + rv = await self._peek_next() + + if rv is missing: + return self._undefined("there is no next item") + + return rv + + def __aiter__(self) -> "AsyncLoopContext": + return self + + async def __anext__(self) -> t.Tuple[t.Any, "AsyncLoopContext"]: + if self._after is not missing: + rv = self._after + self._after = missing + else: + rv = await self._iterator.__anext__() + + self.index0 += 1 + self._before = self._current + self._current = rv + return rv, self + + +class Macro: + """Wraps a macro function.""" + + def __init__( + self, + environment: "Environment", + func: t.Callable[..., str], + name: str, + arguments: t.List[str], + catch_kwargs: bool, + catch_varargs: bool, + caller: bool, + default_autoescape: t.Optional[bool] = None, + ): + self._environment = environment + self._func = func + self._argument_count = len(arguments) + self.name = name + self.arguments = arguments + self.catch_kwargs = catch_kwargs + self.catch_varargs = catch_varargs + self.caller = caller + self.explicit_caller = "caller" in arguments + + if default_autoescape is None: + if callable(environment.autoescape): + default_autoescape = environment.autoescape(None) + else: + default_autoescape = environment.autoescape + + self._default_autoescape = default_autoescape + + @internalcode + @pass_eval_context + def __call__(self, *args: t.Any, **kwargs: t.Any) -> str: + # This requires a bit of explanation, In the past we used to + # decide largely based on compile-time information if a macro is + # safe or unsafe. While there was a volatile mode it was largely + # unused for deciding on escaping. This turns out to be + # problematic for macros because whether a macro is safe depends not + # on the escape mode when it was defined, but rather when it was used. + # + # Because however we export macros from the module system and + # there are historic callers that do not pass an eval context (and + # will continue to not pass one), we need to perform an instance + # check here. + # + # This is considered safe because an eval context is not a valid + # argument to callables otherwise anyway. Worst case here is + # that if no eval context is passed we fall back to the compile + # time autoescape flag. + if args and isinstance(args[0], EvalContext): + autoescape = args[0].autoescape + args = args[1:] + else: + autoescape = self._default_autoescape + + # try to consume the positional arguments + arguments = list(args[: self._argument_count]) + off = len(arguments) + + # For information why this is necessary refer to the handling + # of caller in the `macro_body` handler in the compiler. + found_caller = False + + # if the number of arguments consumed is not the number of + # arguments expected we start filling in keyword arguments + # and defaults. + if off != self._argument_count: + for name in self.arguments[len(arguments) :]: + try: + value = kwargs.pop(name) + except KeyError: + value = missing + if name == "caller": + found_caller = True + arguments.append(value) + else: + found_caller = self.explicit_caller + + # it's important that the order of these arguments does not change + # if not also changed in the compiler's `function_scoping` method. + # the order is caller, keyword arguments, positional arguments! + if self.caller and not found_caller: + caller = kwargs.pop("caller", None) + if caller is None: + caller = self._environment.undefined("No caller defined", name="caller") + arguments.append(caller) + + if self.catch_kwargs: + arguments.append(kwargs) + elif kwargs: + if "caller" in kwargs: + raise TypeError( + f"macro {self.name!r} was invoked with two values for the special" + " caller argument. This is most likely a bug." + ) + raise TypeError( + f"macro {self.name!r} takes no keyword argument {next(iter(kwargs))!r}" + ) + if self.catch_varargs: + arguments.append(args[self._argument_count :]) + elif len(args) > self._argument_count: + raise TypeError( + f"macro {self.name!r} takes not more than" + f" {len(self.arguments)} argument(s)" + ) + + return self._invoke(arguments, autoescape) + + async def _async_invoke(self, arguments: t.List[t.Any], autoescape: bool) -> str: + rv = await self._func(*arguments) # type: ignore + + if autoescape: + return Markup(rv) + + return rv # type: ignore + + def _invoke(self, arguments: t.List[t.Any], autoescape: bool) -> str: + if self._environment.is_async: + return self._async_invoke(arguments, autoescape) # type: ignore + + rv = self._func(*arguments) + + if autoescape: + rv = Markup(rv) + + return rv + + def __repr__(self) -> str: + name = "anonymous" if self.name is None else repr(self.name) + return f"<{type(self).__name__} {name}>" + + +class Undefined: + """The default undefined type. This can be printed, iterated, and treated as + a boolean. Any other operation will raise an :exc:`UndefinedError`. + + >>> foo = Undefined(name='foo') + >>> str(foo) + '' + >>> not foo + True + >>> foo + 42 + Traceback (most recent call last): + ... + jinja2.exceptions.UndefinedError: 'foo' is undefined + """ + + __slots__ = ( + "_undefined_hint", + "_undefined_obj", + "_undefined_name", + "_undefined_exception", + ) + + def __init__( + self, + hint: t.Optional[str] = None, + obj: t.Any = missing, + name: t.Optional[str] = None, + exc: t.Type[TemplateRuntimeError] = UndefinedError, + ) -> None: + self._undefined_hint = hint + self._undefined_obj = obj + self._undefined_name = name + self._undefined_exception = exc + + @property + def _undefined_message(self) -> str: + """Build a message about the undefined value based on how it was + accessed. + """ + if self._undefined_hint: + return self._undefined_hint + + if self._undefined_obj is missing: + return f"{self._undefined_name!r} is undefined" + + if not isinstance(self._undefined_name, str): + return ( + f"{object_type_repr(self._undefined_obj)} has no" + f" element {self._undefined_name!r}" + ) + + return ( + f"{object_type_repr(self._undefined_obj)!r} has no" + f" attribute {self._undefined_name!r}" + ) + + @internalcode + def _fail_with_undefined_error( + self, *args: t.Any, **kwargs: t.Any + ) -> "te.NoReturn": + """Raise an :exc:`UndefinedError` when operations are performed + on the undefined value. + """ + raise self._undefined_exception(self._undefined_message) + + @internalcode + def __getattr__(self, name: str) -> t.Any: + # Raise AttributeError on requests for names that appear to be unimplemented + # dunder methods to keep Python's internal protocol probing behaviors working + # properly in cases where another exception type could cause unexpected or + # difficult-to-diagnose failures. + if name[:2] == "__" and name[-2:] == "__": + raise AttributeError(name) + + return self._fail_with_undefined_error() + + __add__ = __radd__ = __sub__ = __rsub__ = _fail_with_undefined_error + __mul__ = __rmul__ = __div__ = __rdiv__ = _fail_with_undefined_error + __truediv__ = __rtruediv__ = _fail_with_undefined_error + __floordiv__ = __rfloordiv__ = _fail_with_undefined_error + __mod__ = __rmod__ = _fail_with_undefined_error + __pos__ = __neg__ = _fail_with_undefined_error + __call__ = __getitem__ = _fail_with_undefined_error + __lt__ = __le__ = __gt__ = __ge__ = _fail_with_undefined_error + __int__ = __float__ = __complex__ = _fail_with_undefined_error + __pow__ = __rpow__ = _fail_with_undefined_error + + def __eq__(self, other: t.Any) -> bool: + return type(self) is type(other) + + def __ne__(self, other: t.Any) -> bool: + return not self.__eq__(other) + + def __hash__(self) -> int: + return id(type(self)) + + def __str__(self) -> str: + return "" + + def __len__(self) -> int: + return 0 + + def __iter__(self) -> t.Iterator[t.Any]: + yield from () + + async def __aiter__(self) -> t.AsyncIterator[t.Any]: + for _ in (): + yield + + def __bool__(self) -> bool: + return False + + def __repr__(self) -> str: + return "Undefined" + + +def make_logging_undefined( + logger: t.Optional["logging.Logger"] = None, base: t.Type[Undefined] = Undefined +) -> t.Type[Undefined]: + """Given a logger object this returns a new undefined class that will + log certain failures. It will log iterations and printing. If no + logger is given a default logger is created. + + Example:: + + logger = logging.getLogger(__name__) + LoggingUndefined = make_logging_undefined( + logger=logger, + base=Undefined + ) + + .. versionadded:: 2.8 + + :param logger: the logger to use. If not provided, a default logger + is created. + :param base: the base class to add logging functionality to. This + defaults to :class:`Undefined`. + """ + if logger is None: + import logging + + logger = logging.getLogger(__name__) + logger.addHandler(logging.StreamHandler(sys.stderr)) + + def _log_message(undef: Undefined) -> None: + logger.warning("Template variable warning: %s", undef._undefined_message) + + class LoggingUndefined(base): # type: ignore + __slots__ = () + + def _fail_with_undefined_error( # type: ignore + self, *args: t.Any, **kwargs: t.Any + ) -> "te.NoReturn": + try: + super()._fail_with_undefined_error(*args, **kwargs) + except self._undefined_exception as e: + logger.error("Template variable error: %s", e) # type: ignore + raise e + + def __str__(self) -> str: + _log_message(self) + return super().__str__() # type: ignore + + def __iter__(self) -> t.Iterator[t.Any]: + _log_message(self) + return super().__iter__() # type: ignore + + def __bool__(self) -> bool: + _log_message(self) + return super().__bool__() # type: ignore + + return LoggingUndefined + + +class ChainableUndefined(Undefined): + """An undefined that is chainable, where both ``__getattr__`` and + ``__getitem__`` return itself rather than raising an + :exc:`UndefinedError`. + + >>> foo = ChainableUndefined(name='foo') + >>> str(foo.bar['baz']) + '' + >>> foo.bar['baz'] + 42 + Traceback (most recent call last): + ... + jinja2.exceptions.UndefinedError: 'foo' is undefined + + .. versionadded:: 2.11.0 + """ + + __slots__ = () + + def __html__(self) -> str: + return str(self) + + def __getattr__(self, name: str) -> "ChainableUndefined": + # Raise AttributeError on requests for names that appear to be unimplemented + # dunder methods to avoid confusing Python with truthy non-method objects that + # do not implement the protocol being probed for. e.g., copy.copy(Undefined()) + # fails spectacularly if getattr(Undefined(), '__setstate__') returns an + # Undefined object instead of raising AttributeError to signal that it does not + # support that style of object initialization. + if name[:2] == "__" and name[-2:] == "__": + raise AttributeError(name) + + return self + + def __getitem__(self, _name: str) -> "ChainableUndefined": # type: ignore[override] + return self + + +class DebugUndefined(Undefined): + """An undefined that returns the debug info when printed. + + >>> foo = DebugUndefined(name='foo') + >>> str(foo) + '{{ foo }}' + >>> not foo + True + >>> foo + 42 + Traceback (most recent call last): + ... + jinja2.exceptions.UndefinedError: 'foo' is undefined + """ + + __slots__ = () + + def __str__(self) -> str: + if self._undefined_hint: + message = f"undefined value printed: {self._undefined_hint}" + + elif self._undefined_obj is missing: + message = self._undefined_name # type: ignore + + else: + message = ( + f"no such element: {object_type_repr(self._undefined_obj)}" + f"[{self._undefined_name!r}]" + ) + + return f"{{{{ {message} }}}}" + + +class StrictUndefined(Undefined): + """An undefined that barks on print and iteration as well as boolean + tests and all kinds of comparisons. In other words: you can do nothing + with it except checking if it's defined using the `defined` test. + + >>> foo = StrictUndefined(name='foo') + >>> str(foo) + Traceback (most recent call last): + ... + jinja2.exceptions.UndefinedError: 'foo' is undefined + >>> not foo + Traceback (most recent call last): + ... + jinja2.exceptions.UndefinedError: 'foo' is undefined + >>> foo + 42 + Traceback (most recent call last): + ... + jinja2.exceptions.UndefinedError: 'foo' is undefined + """ + + __slots__ = () + __iter__ = __str__ = __len__ = Undefined._fail_with_undefined_error + __eq__ = __ne__ = __bool__ = __hash__ = Undefined._fail_with_undefined_error + __contains__ = Undefined._fail_with_undefined_error diff --git a/venv/Lib/site-packages/jinja2/sandbox.py b/venv/Lib/site-packages/jinja2/sandbox.py new file mode 100644 index 0000000..9c9dae2 --- /dev/null +++ b/venv/Lib/site-packages/jinja2/sandbox.py @@ -0,0 +1,436 @@ +"""A sandbox layer that ensures unsafe operations cannot be performed. +Useful when the template itself comes from an untrusted source. +""" + +import operator +import types +import typing as t +from _string import formatter_field_name_split # type: ignore +from collections import abc +from collections import deque +from functools import update_wrapper +from string import Formatter + +from markupsafe import EscapeFormatter +from markupsafe import Markup + +from .environment import Environment +from .exceptions import SecurityError +from .runtime import Context +from .runtime import Undefined + +F = t.TypeVar("F", bound=t.Callable[..., t.Any]) + +#: maximum number of items a range may produce +MAX_RANGE = 100000 + +#: Unsafe function attributes. +UNSAFE_FUNCTION_ATTRIBUTES: t.Set[str] = set() + +#: Unsafe method attributes. Function attributes are unsafe for methods too. +UNSAFE_METHOD_ATTRIBUTES: t.Set[str] = set() + +#: unsafe generator attributes. +UNSAFE_GENERATOR_ATTRIBUTES = {"gi_frame", "gi_code"} + +#: unsafe attributes on coroutines +UNSAFE_COROUTINE_ATTRIBUTES = {"cr_frame", "cr_code"} + +#: unsafe attributes on async generators +UNSAFE_ASYNC_GENERATOR_ATTRIBUTES = {"ag_code", "ag_frame"} + +_mutable_spec: t.Tuple[t.Tuple[t.Type[t.Any], t.FrozenSet[str]], ...] = ( + ( + abc.MutableSet, + frozenset( + [ + "add", + "clear", + "difference_update", + "discard", + "pop", + "remove", + "symmetric_difference_update", + "update", + ] + ), + ), + ( + abc.MutableMapping, + frozenset(["clear", "pop", "popitem", "setdefault", "update"]), + ), + ( + abc.MutableSequence, + frozenset( + ["append", "clear", "pop", "reverse", "insert", "sort", "extend", "remove"] + ), + ), + ( + deque, + frozenset( + [ + "append", + "appendleft", + "clear", + "extend", + "extendleft", + "pop", + "popleft", + "remove", + "rotate", + ] + ), + ), +) + + +def safe_range(*args: int) -> range: + """A range that can't generate ranges with a length of more than + MAX_RANGE items. + """ + rng = range(*args) + + if len(rng) > MAX_RANGE: + raise OverflowError( + "Range too big. The sandbox blocks ranges larger than" + f" MAX_RANGE ({MAX_RANGE})." + ) + + return rng + + +def unsafe(f: F) -> F: + """Marks a function or method as unsafe. + + .. code-block: python + + @unsafe + def delete(self): + pass + """ + f.unsafe_callable = True # type: ignore + return f + + +def is_internal_attribute(obj: t.Any, attr: str) -> bool: + """Test if the attribute given is an internal python attribute. For + example this function returns `True` for the `func_code` attribute of + python objects. This is useful if the environment method + :meth:`~SandboxedEnvironment.is_safe_attribute` is overridden. + + >>> from jinja2.sandbox import is_internal_attribute + >>> is_internal_attribute(str, "mro") + True + >>> is_internal_attribute(str, "upper") + False + """ + if isinstance(obj, types.FunctionType): + if attr in UNSAFE_FUNCTION_ATTRIBUTES: + return True + elif isinstance(obj, types.MethodType): + if attr in UNSAFE_FUNCTION_ATTRIBUTES or attr in UNSAFE_METHOD_ATTRIBUTES: + return True + elif isinstance(obj, type): + if attr == "mro": + return True + elif isinstance(obj, (types.CodeType, types.TracebackType, types.FrameType)): + return True + elif isinstance(obj, types.GeneratorType): + if attr in UNSAFE_GENERATOR_ATTRIBUTES: + return True + elif hasattr(types, "CoroutineType") and isinstance(obj, types.CoroutineType): + if attr in UNSAFE_COROUTINE_ATTRIBUTES: + return True + elif hasattr(types, "AsyncGeneratorType") and isinstance( + obj, types.AsyncGeneratorType + ): + if attr in UNSAFE_ASYNC_GENERATOR_ATTRIBUTES: + return True + return attr.startswith("__") + + +def modifies_known_mutable(obj: t.Any, attr: str) -> bool: + """This function checks if an attribute on a builtin mutable object + (list, dict, set or deque) or the corresponding ABCs would modify it + if called. + + >>> modifies_known_mutable({}, "clear") + True + >>> modifies_known_mutable({}, "keys") + False + >>> modifies_known_mutable([], "append") + True + >>> modifies_known_mutable([], "index") + False + + If called with an unsupported object, ``False`` is returned. + + >>> modifies_known_mutable("foo", "upper") + False + """ + for typespec, unsafe in _mutable_spec: + if isinstance(obj, typespec): + return attr in unsafe + return False + + +class SandboxedEnvironment(Environment): + """The sandboxed environment. It works like the regular environment but + tells the compiler to generate sandboxed code. Additionally subclasses of + this environment may override the methods that tell the runtime what + attributes or functions are safe to access. + + If the template tries to access insecure code a :exc:`SecurityError` is + raised. However also other exceptions may occur during the rendering so + the caller has to ensure that all exceptions are caught. + """ + + sandboxed = True + + #: default callback table for the binary operators. A copy of this is + #: available on each instance of a sandboxed environment as + #: :attr:`binop_table` + default_binop_table: t.Dict[str, t.Callable[[t.Any, t.Any], t.Any]] = { + "+": operator.add, + "-": operator.sub, + "*": operator.mul, + "/": operator.truediv, + "//": operator.floordiv, + "**": operator.pow, + "%": operator.mod, + } + + #: default callback table for the unary operators. A copy of this is + #: available on each instance of a sandboxed environment as + #: :attr:`unop_table` + default_unop_table: t.Dict[str, t.Callable[[t.Any], t.Any]] = { + "+": operator.pos, + "-": operator.neg, + } + + #: a set of binary operators that should be intercepted. Each operator + #: that is added to this set (empty by default) is delegated to the + #: :meth:`call_binop` method that will perform the operator. The default + #: operator callback is specified by :attr:`binop_table`. + #: + #: The following binary operators are interceptable: + #: ``//``, ``%``, ``+``, ``*``, ``-``, ``/``, and ``**`` + #: + #: The default operation form the operator table corresponds to the + #: builtin function. Intercepted calls are always slower than the native + #: operator call, so make sure only to intercept the ones you are + #: interested in. + #: + #: .. versionadded:: 2.6 + intercepted_binops: t.FrozenSet[str] = frozenset() + + #: a set of unary operators that should be intercepted. Each operator + #: that is added to this set (empty by default) is delegated to the + #: :meth:`call_unop` method that will perform the operator. The default + #: operator callback is specified by :attr:`unop_table`. + #: + #: The following unary operators are interceptable: ``+``, ``-`` + #: + #: The default operation form the operator table corresponds to the + #: builtin function. Intercepted calls are always slower than the native + #: operator call, so make sure only to intercept the ones you are + #: interested in. + #: + #: .. versionadded:: 2.6 + intercepted_unops: t.FrozenSet[str] = frozenset() + + def __init__(self, *args: t.Any, **kwargs: t.Any) -> None: + super().__init__(*args, **kwargs) + self.globals["range"] = safe_range + self.binop_table = self.default_binop_table.copy() + self.unop_table = self.default_unop_table.copy() + + def is_safe_attribute(self, obj: t.Any, attr: str, value: t.Any) -> bool: + """The sandboxed environment will call this method to check if the + attribute of an object is safe to access. Per default all attributes + starting with an underscore are considered private as well as the + special attributes of internal python objects as returned by the + :func:`is_internal_attribute` function. + """ + return not (attr.startswith("_") or is_internal_attribute(obj, attr)) + + def is_safe_callable(self, obj: t.Any) -> bool: + """Check if an object is safely callable. By default callables + are considered safe unless decorated with :func:`unsafe`. + + This also recognizes the Django convention of setting + ``func.alters_data = True``. + """ + return not ( + getattr(obj, "unsafe_callable", False) or getattr(obj, "alters_data", False) + ) + + def call_binop( + self, context: Context, operator: str, left: t.Any, right: t.Any + ) -> t.Any: + """For intercepted binary operator calls (:meth:`intercepted_binops`) + this function is executed instead of the builtin operator. This can + be used to fine tune the behavior of certain operators. + + .. versionadded:: 2.6 + """ + return self.binop_table[operator](left, right) + + def call_unop(self, context: Context, operator: str, arg: t.Any) -> t.Any: + """For intercepted unary operator calls (:meth:`intercepted_unops`) + this function is executed instead of the builtin operator. This can + be used to fine tune the behavior of certain operators. + + .. versionadded:: 2.6 + """ + return self.unop_table[operator](arg) + + def getitem( + self, obj: t.Any, argument: t.Union[str, t.Any] + ) -> t.Union[t.Any, Undefined]: + """Subscribe an object from sandboxed code.""" + try: + return obj[argument] + except (TypeError, LookupError): + if isinstance(argument, str): + try: + attr = str(argument) + except Exception: + pass + else: + try: + value = getattr(obj, attr) + except AttributeError: + pass + else: + fmt = self.wrap_str_format(value) + if fmt is not None: + return fmt + if self.is_safe_attribute(obj, argument, value): + return value + return self.unsafe_undefined(obj, argument) + return self.undefined(obj=obj, name=argument) + + def getattr(self, obj: t.Any, attribute: str) -> t.Union[t.Any, Undefined]: + """Subscribe an object from sandboxed code and prefer the + attribute. The attribute passed *must* be a bytestring. + """ + try: + value = getattr(obj, attribute) + except AttributeError: + try: + return obj[attribute] + except (TypeError, LookupError): + pass + else: + fmt = self.wrap_str_format(value) + if fmt is not None: + return fmt + if self.is_safe_attribute(obj, attribute, value): + return value + return self.unsafe_undefined(obj, attribute) + return self.undefined(obj=obj, name=attribute) + + def unsafe_undefined(self, obj: t.Any, attribute: str) -> Undefined: + """Return an undefined object for unsafe attributes.""" + return self.undefined( + f"access to attribute {attribute!r} of" + f" {type(obj).__name__!r} object is unsafe.", + name=attribute, + obj=obj, + exc=SecurityError, + ) + + def wrap_str_format(self, value: t.Any) -> t.Optional[t.Callable[..., str]]: + """If the given value is a ``str.format`` or ``str.format_map`` method, + return a new function than handles sandboxing. This is done at access + rather than in :meth:`call`, so that calls made without ``call`` are + also sandboxed. + """ + if not isinstance( + value, (types.MethodType, types.BuiltinMethodType) + ) or value.__name__ not in ("format", "format_map"): + return None + + f_self: t.Any = value.__self__ + + if not isinstance(f_self, str): + return None + + str_type: t.Type[str] = type(f_self) + is_format_map = value.__name__ == "format_map" + formatter: SandboxedFormatter + + if isinstance(f_self, Markup): + formatter = SandboxedEscapeFormatter(self, escape=f_self.escape) + else: + formatter = SandboxedFormatter(self) + + vformat = formatter.vformat + + def wrapper(*args: t.Any, **kwargs: t.Any) -> str: + if is_format_map: + if kwargs: + raise TypeError("format_map() takes no keyword arguments") + + if len(args) != 1: + raise TypeError( + f"format_map() takes exactly one argument ({len(args)} given)" + ) + + kwargs = args[0] + args = () + + return str_type(vformat(f_self, args, kwargs)) + + return update_wrapper(wrapper, value) + + def call( + __self, # noqa: B902 + __context: Context, + __obj: t.Any, + *args: t.Any, + **kwargs: t.Any, + ) -> t.Any: + """Call an object from sandboxed code.""" + + # the double prefixes are to avoid double keyword argument + # errors when proxying the call. + if not __self.is_safe_callable(__obj): + raise SecurityError(f"{__obj!r} is not safely callable") + return __context.call(__obj, *args, **kwargs) + + +class ImmutableSandboxedEnvironment(SandboxedEnvironment): + """Works exactly like the regular `SandboxedEnvironment` but does not + permit modifications on the builtin mutable objects `list`, `set`, and + `dict` by using the :func:`modifies_known_mutable` function. + """ + + def is_safe_attribute(self, obj: t.Any, attr: str, value: t.Any) -> bool: + if not super().is_safe_attribute(obj, attr, value): + return False + + return not modifies_known_mutable(obj, attr) + + +class SandboxedFormatter(Formatter): + def __init__(self, env: Environment, **kwargs: t.Any) -> None: + self._env = env + super().__init__(**kwargs) + + def get_field( + self, field_name: str, args: t.Sequence[t.Any], kwargs: t.Mapping[str, t.Any] + ) -> t.Tuple[t.Any, str]: + first, rest = formatter_field_name_split(field_name) + obj = self.get_value(first, args, kwargs) + for is_attr, i in rest: + if is_attr: + obj = self._env.getattr(obj, i) + else: + obj = self._env.getitem(obj, i) + return obj, first + + +class SandboxedEscapeFormatter(SandboxedFormatter, EscapeFormatter): + pass diff --git a/venv/Lib/site-packages/jinja2/tests.py b/venv/Lib/site-packages/jinja2/tests.py new file mode 100644 index 0000000..1a59e37 --- /dev/null +++ b/venv/Lib/site-packages/jinja2/tests.py @@ -0,0 +1,256 @@ +"""Built-in template tests used with the ``is`` operator.""" + +import operator +import typing as t +from collections import abc +from numbers import Number + +from .runtime import Undefined +from .utils import pass_environment + +if t.TYPE_CHECKING: + from .environment import Environment + + +def test_odd(value: int) -> bool: + """Return true if the variable is odd.""" + return value % 2 == 1 + + +def test_even(value: int) -> bool: + """Return true if the variable is even.""" + return value % 2 == 0 + + +def test_divisibleby(value: int, num: int) -> bool: + """Check if a variable is divisible by a number.""" + return value % num == 0 + + +def test_defined(value: t.Any) -> bool: + """Return true if the variable is defined: + + .. sourcecode:: jinja + + {% if variable is defined %} + value of variable: {{ variable }} + {% else %} + variable is not defined + {% endif %} + + See the :func:`default` filter for a simple way to set undefined + variables. + """ + return not isinstance(value, Undefined) + + +def test_undefined(value: t.Any) -> bool: + """Like :func:`defined` but the other way round.""" + return isinstance(value, Undefined) + + +@pass_environment +def test_filter(env: "Environment", value: str) -> bool: + """Check if a filter exists by name. Useful if a filter may be + optionally available. + + .. code-block:: jinja + + {% if 'markdown' is filter %} + {{ value | markdown }} + {% else %} + {{ value }} + {% endif %} + + .. versionadded:: 3.0 + """ + return value in env.filters + + +@pass_environment +def test_test(env: "Environment", value: str) -> bool: + """Check if a test exists by name. Useful if a test may be + optionally available. + + .. code-block:: jinja + + {% if 'loud' is test %} + {% if value is loud %} + {{ value|upper }} + {% else %} + {{ value|lower }} + {% endif %} + {% else %} + {{ value }} + {% endif %} + + .. versionadded:: 3.0 + """ + return value in env.tests + + +def test_none(value: t.Any) -> bool: + """Return true if the variable is none.""" + return value is None + + +def test_boolean(value: t.Any) -> bool: + """Return true if the object is a boolean value. + + .. versionadded:: 2.11 + """ + return value is True or value is False + + +def test_false(value: t.Any) -> bool: + """Return true if the object is False. + + .. versionadded:: 2.11 + """ + return value is False + + +def test_true(value: t.Any) -> bool: + """Return true if the object is True. + + .. versionadded:: 2.11 + """ + return value is True + + +# NOTE: The existing 'number' test matches booleans and floats +def test_integer(value: t.Any) -> bool: + """Return true if the object is an integer. + + .. versionadded:: 2.11 + """ + return isinstance(value, int) and value is not True and value is not False + + +# NOTE: The existing 'number' test matches booleans and integers +def test_float(value: t.Any) -> bool: + """Return true if the object is a float. + + .. versionadded:: 2.11 + """ + return isinstance(value, float) + + +def test_lower(value: str) -> bool: + """Return true if the variable is lowercased.""" + return str(value).islower() + + +def test_upper(value: str) -> bool: + """Return true if the variable is uppercased.""" + return str(value).isupper() + + +def test_string(value: t.Any) -> bool: + """Return true if the object is a string.""" + return isinstance(value, str) + + +def test_mapping(value: t.Any) -> bool: + """Return true if the object is a mapping (dict etc.). + + .. versionadded:: 2.6 + """ + return isinstance(value, abc.Mapping) + + +def test_number(value: t.Any) -> bool: + """Return true if the variable is a number.""" + return isinstance(value, Number) + + +def test_sequence(value: t.Any) -> bool: + """Return true if the variable is a sequence. Sequences are variables + that are iterable. + """ + try: + len(value) + value.__getitem__ # noqa B018 + except Exception: + return False + + return True + + +def test_sameas(value: t.Any, other: t.Any) -> bool: + """Check if an object points to the same memory address than another + object: + + .. sourcecode:: jinja + + {% if foo.attribute is sameas false %} + the foo attribute really is the `False` singleton + {% endif %} + """ + return value is other + + +def test_iterable(value: t.Any) -> bool: + """Check if it's possible to iterate over an object.""" + try: + iter(value) + except TypeError: + return False + + return True + + +def test_escaped(value: t.Any) -> bool: + """Check if the value is escaped.""" + return hasattr(value, "__html__") + + +def test_in(value: t.Any, seq: t.Container[t.Any]) -> bool: + """Check if value is in seq. + + .. versionadded:: 2.10 + """ + return value in seq + + +TESTS = { + "odd": test_odd, + "even": test_even, + "divisibleby": test_divisibleby, + "defined": test_defined, + "undefined": test_undefined, + "filter": test_filter, + "test": test_test, + "none": test_none, + "boolean": test_boolean, + "false": test_false, + "true": test_true, + "integer": test_integer, + "float": test_float, + "lower": test_lower, + "upper": test_upper, + "string": test_string, + "mapping": test_mapping, + "number": test_number, + "sequence": test_sequence, + "iterable": test_iterable, + "callable": callable, + "sameas": test_sameas, + "escaped": test_escaped, + "in": test_in, + "==": operator.eq, + "eq": operator.eq, + "equalto": operator.eq, + "!=": operator.ne, + "ne": operator.ne, + ">": operator.gt, + "gt": operator.gt, + "greaterthan": operator.gt, + "ge": operator.ge, + ">=": operator.ge, + "<": operator.lt, + "lt": operator.lt, + "lessthan": operator.lt, + "<=": operator.le, + "le": operator.le, +} diff --git a/venv/Lib/site-packages/jinja2/utils.py b/venv/Lib/site-packages/jinja2/utils.py new file mode 100644 index 0000000..7c92262 --- /dev/null +++ b/venv/Lib/site-packages/jinja2/utils.py @@ -0,0 +1,766 @@ +import enum +import json +import os +import re +import typing as t +from collections import abc +from collections import deque +from random import choice +from random import randrange +from threading import Lock +from types import CodeType +from urllib.parse import quote_from_bytes + +import markupsafe + +if t.TYPE_CHECKING: + import typing_extensions as te + +F = t.TypeVar("F", bound=t.Callable[..., t.Any]) + + +class _MissingType: + def __repr__(self) -> str: + return "missing" + + def __reduce__(self) -> str: + return "missing" + + +missing: t.Any = _MissingType() +"""Special singleton representing missing values for the runtime.""" + +internal_code: t.MutableSet[CodeType] = set() + +concat = "".join + + +def pass_context(f: F) -> F: + """Pass the :class:`~jinja2.runtime.Context` as the first argument + to the decorated function when called while rendering a template. + + Can be used on functions, filters, and tests. + + If only ``Context.eval_context`` is needed, use + :func:`pass_eval_context`. If only ``Context.environment`` is + needed, use :func:`pass_environment`. + + .. versionadded:: 3.0.0 + Replaces ``contextfunction`` and ``contextfilter``. + """ + f.jinja_pass_arg = _PassArg.context # type: ignore + return f + + +def pass_eval_context(f: F) -> F: + """Pass the :class:`~jinja2.nodes.EvalContext` as the first argument + to the decorated function when called while rendering a template. + See :ref:`eval-context`. + + Can be used on functions, filters, and tests. + + If only ``EvalContext.environment`` is needed, use + :func:`pass_environment`. + + .. versionadded:: 3.0.0 + Replaces ``evalcontextfunction`` and ``evalcontextfilter``. + """ + f.jinja_pass_arg = _PassArg.eval_context # type: ignore + return f + + +def pass_environment(f: F) -> F: + """Pass the :class:`~jinja2.Environment` as the first argument to + the decorated function when called while rendering a template. + + Can be used on functions, filters, and tests. + + .. versionadded:: 3.0.0 + Replaces ``environmentfunction`` and ``environmentfilter``. + """ + f.jinja_pass_arg = _PassArg.environment # type: ignore + return f + + +class _PassArg(enum.Enum): + context = enum.auto() + eval_context = enum.auto() + environment = enum.auto() + + @classmethod + def from_obj(cls, obj: F) -> t.Optional["_PassArg"]: + if hasattr(obj, "jinja_pass_arg"): + return obj.jinja_pass_arg # type: ignore + + return None + + +def internalcode(f: F) -> F: + """Marks the function as internally used""" + internal_code.add(f.__code__) + return f + + +def is_undefined(obj: t.Any) -> bool: + """Check if the object passed is undefined. This does nothing more than + performing an instance check against :class:`Undefined` but looks nicer. + This can be used for custom filters or tests that want to react to + undefined variables. For example a custom default filter can look like + this:: + + def default(var, default=''): + if is_undefined(var): + return default + return var + """ + from .runtime import Undefined + + return isinstance(obj, Undefined) + + +def consume(iterable: t.Iterable[t.Any]) -> None: + """Consumes an iterable without doing anything with it.""" + for _ in iterable: + pass + + +def clear_caches() -> None: + """Jinja keeps internal caches for environments and lexers. These are + used so that Jinja doesn't have to recreate environments and lexers all + the time. Normally you don't have to care about that but if you are + measuring memory consumption you may want to clean the caches. + """ + from .environment import get_spontaneous_environment + from .lexer import _lexer_cache + + get_spontaneous_environment.cache_clear() + _lexer_cache.clear() + + +def import_string(import_name: str, silent: bool = False) -> t.Any: + """Imports an object based on a string. This is useful if you want to + use import paths as endpoints or something similar. An import path can + be specified either in dotted notation (``xml.sax.saxutils.escape``) + or with a colon as object delimiter (``xml.sax.saxutils:escape``). + + If the `silent` is True the return value will be `None` if the import + fails. + + :return: imported object + """ + try: + if ":" in import_name: + module, obj = import_name.split(":", 1) + elif "." in import_name: + module, _, obj = import_name.rpartition(".") + else: + return __import__(import_name) + return getattr(__import__(module, None, None, [obj]), obj) + except (ImportError, AttributeError): + if not silent: + raise + + +def open_if_exists(filename: str, mode: str = "rb") -> t.Optional[t.IO[t.Any]]: + """Returns a file descriptor for the filename if that file exists, + otherwise ``None``. + """ + if not os.path.isfile(filename): + return None + + return open(filename, mode) + + +def object_type_repr(obj: t.Any) -> str: + """Returns the name of the object's type. For some recognized + singletons the name of the object is returned instead. (For + example for `None` and `Ellipsis`). + """ + if obj is None: + return "None" + elif obj is Ellipsis: + return "Ellipsis" + + cls = type(obj) + + if cls.__module__ == "builtins": + return f"{cls.__name__} object" + + return f"{cls.__module__}.{cls.__name__} object" + + +def pformat(obj: t.Any) -> str: + """Format an object using :func:`pprint.pformat`.""" + from pprint import pformat + + return pformat(obj) + + +_http_re = re.compile( + r""" + ^ + ( + (https?://|www\.) # scheme or www + (([\w%-]+\.)+)? # subdomain + ( + [a-z]{2,63} # basic tld + | + xn--[\w%]{2,59} # idna tld + ) + | + ([\w%-]{2,63}\.)+ # basic domain + (com|net|int|edu|gov|org|info|mil) # basic tld + | + (https?://) # scheme + ( + (([\d]{1,3})(\.[\d]{1,3}){3}) # IPv4 + | + (\[([\da-f]{0,4}:){2}([\da-f]{0,4}:?){1,6}]) # IPv6 + ) + ) + (?::[\d]{1,5})? # port + (?:[/?#]\S*)? # path, query, and fragment + $ + """, + re.IGNORECASE | re.VERBOSE, +) +_email_re = re.compile(r"^\S+@\w[\w.-]*\.\w+$") + + +def urlize( + text: str, + trim_url_limit: t.Optional[int] = None, + rel: t.Optional[str] = None, + target: t.Optional[str] = None, + extra_schemes: t.Optional[t.Iterable[str]] = None, +) -> str: + """Convert URLs in text into clickable links. + + This may not recognize links in some situations. Usually, a more + comprehensive formatter, such as a Markdown library, is a better + choice. + + Works on ``http://``, ``https://``, ``www.``, ``mailto:``, and email + addresses. Links with trailing punctuation (periods, commas, closing + parentheses) and leading punctuation (opening parentheses) are + recognized excluding the punctuation. Email addresses that include + header fields are not recognized (for example, + ``mailto:address@example.com?cc=copy@example.com``). + + :param text: Original text containing URLs to link. + :param trim_url_limit: Shorten displayed URL values to this length. + :param target: Add the ``target`` attribute to links. + :param rel: Add the ``rel`` attribute to links. + :param extra_schemes: Recognize URLs that start with these schemes + in addition to the default behavior. + + .. versionchanged:: 3.0 + The ``extra_schemes`` parameter was added. + + .. versionchanged:: 3.0 + Generate ``https://`` links for URLs without a scheme. + + .. versionchanged:: 3.0 + The parsing rules were updated. Recognize email addresses with + or without the ``mailto:`` scheme. Validate IP addresses. Ignore + parentheses and brackets in more cases. + """ + if trim_url_limit is not None: + + def trim_url(x: str) -> str: + if len(x) > trim_url_limit: + return f"{x[:trim_url_limit]}..." + + return x + + else: + + def trim_url(x: str) -> str: + return x + + words = re.split(r"(\s+)", str(markupsafe.escape(text))) + rel_attr = f' rel="{markupsafe.escape(rel)}"' if rel else "" + target_attr = f' target="{markupsafe.escape(target)}"' if target else "" + + for i, word in enumerate(words): + head, middle, tail = "", word, "" + match = re.match(r"^([(<]|<)+", middle) + + if match: + head = match.group() + middle = middle[match.end() :] + + # Unlike lead, which is anchored to the start of the string, + # need to check that the string ends with any of the characters + # before trying to match all of them, to avoid backtracking. + if middle.endswith((")", ">", ".", ",", "\n", ">")): + match = re.search(r"([)>.,\n]|>)+$", middle) + + if match: + tail = match.group() + middle = middle[: match.start()] + + # Prefer balancing parentheses in URLs instead of ignoring a + # trailing character. + for start_char, end_char in ("(", ")"), ("<", ">"), ("<", ">"): + start_count = middle.count(start_char) + + if start_count <= middle.count(end_char): + # Balanced, or lighter on the left + continue + + # Move as many as possible from the tail to balance + for _ in range(min(start_count, tail.count(end_char))): + end_index = tail.index(end_char) + len(end_char) + # Move anything in the tail before the end char too + middle += tail[:end_index] + tail = tail[end_index:] + + if _http_re.match(middle): + if middle.startswith("https://") or middle.startswith("http://"): + middle = ( + f'{trim_url(middle)}' + ) + else: + middle = ( + f'' + f"{trim_url(middle)}" + ) + + elif middle.startswith("mailto:") and _email_re.match(middle[7:]): + middle = f'{middle[7:]}' + + elif ( + "@" in middle + and not middle.startswith("www.") + # ignore values like `@a@b` + and not middle.startswith("@") + and ":" not in middle + and _email_re.match(middle) + ): + middle = f'{middle}' + + elif extra_schemes is not None: + for scheme in extra_schemes: + if middle != scheme and middle.startswith(scheme): + middle = f'{middle}' + + words[i] = f"{head}{middle}{tail}" + + return "".join(words) + + +def generate_lorem_ipsum( + n: int = 5, html: bool = True, min: int = 20, max: int = 100 +) -> str: + """Generate some lorem ipsum for the template.""" + from .constants import LOREM_IPSUM_WORDS + + words = LOREM_IPSUM_WORDS.split() + result = [] + + for _ in range(n): + next_capitalized = True + last_comma = last_fullstop = 0 + word = None + last = None + p = [] + + # each paragraph contains out of 20 to 100 words. + for idx, _ in enumerate(range(randrange(min, max))): + while True: + word = choice(words) + if word != last: + last = word + break + if next_capitalized: + word = word.capitalize() + next_capitalized = False + # add commas + if idx - randrange(3, 8) > last_comma: + last_comma = idx + last_fullstop += 2 + word += "," + # add end of sentences + if idx - randrange(10, 20) > last_fullstop: + last_comma = last_fullstop = idx + word += "." + next_capitalized = True + p.append(word) + + # ensure that the paragraph ends with a dot. + p_str = " ".join(p) + + if p_str.endswith(","): + p_str = p_str[:-1] + "." + elif not p_str.endswith("."): + p_str += "." + + result.append(p_str) + + if not html: + return "\n\n".join(result) + return markupsafe.Markup( + "\n".join(f"

{markupsafe.escape(x)}

" for x in result) + ) + + +def url_quote(obj: t.Any, charset: str = "utf-8", for_qs: bool = False) -> str: + """Quote a string for use in a URL using the given charset. + + :param obj: String or bytes to quote. Other types are converted to + string then encoded to bytes using the given charset. + :param charset: Encode text to bytes using this charset. + :param for_qs: Quote "/" and use "+" for spaces. + """ + if not isinstance(obj, bytes): + if not isinstance(obj, str): + obj = str(obj) + + obj = obj.encode(charset) + + safe = b"" if for_qs else b"/" + rv = quote_from_bytes(obj, safe) + + if for_qs: + rv = rv.replace("%20", "+") + + return rv + + +@abc.MutableMapping.register +class LRUCache: + """A simple LRU Cache implementation.""" + + # this is fast for small capacities (something below 1000) but doesn't + # scale. But as long as it's only used as storage for templates this + # won't do any harm. + + def __init__(self, capacity: int) -> None: + self.capacity = capacity + self._mapping: t.Dict[t.Any, t.Any] = {} + self._queue: te.Deque[t.Any] = deque() + self._postinit() + + def _postinit(self) -> None: + # alias all queue methods for faster lookup + self._popleft = self._queue.popleft + self._pop = self._queue.pop + self._remove = self._queue.remove + self._wlock = Lock() + self._append = self._queue.append + + def __getstate__(self) -> t.Mapping[str, t.Any]: + return { + "capacity": self.capacity, + "_mapping": self._mapping, + "_queue": self._queue, + } + + def __setstate__(self, d: t.Mapping[str, t.Any]) -> None: + self.__dict__.update(d) + self._postinit() + + def __getnewargs__(self) -> t.Tuple[t.Any, ...]: + return (self.capacity,) + + def copy(self) -> "te.Self": + """Return a shallow copy of the instance.""" + rv = self.__class__(self.capacity) + rv._mapping.update(self._mapping) + rv._queue.extend(self._queue) + return rv + + def get(self, key: t.Any, default: t.Any = None) -> t.Any: + """Return an item from the cache dict or `default`""" + try: + return self[key] + except KeyError: + return default + + def setdefault(self, key: t.Any, default: t.Any = None) -> t.Any: + """Set `default` if the key is not in the cache otherwise + leave unchanged. Return the value of this key. + """ + try: + return self[key] + except KeyError: + self[key] = default + return default + + def clear(self) -> None: + """Clear the cache.""" + with self._wlock: + self._mapping.clear() + self._queue.clear() + + def __contains__(self, key: t.Any) -> bool: + """Check if a key exists in this cache.""" + return key in self._mapping + + def __len__(self) -> int: + """Return the current size of the cache.""" + return len(self._mapping) + + def __repr__(self) -> str: + return f"<{type(self).__name__} {self._mapping!r}>" + + def __getitem__(self, key: t.Any) -> t.Any: + """Get an item from the cache. Moves the item up so that it has the + highest priority then. + + Raise a `KeyError` if it does not exist. + """ + with self._wlock: + rv = self._mapping[key] + + if self._queue[-1] != key: + try: + self._remove(key) + except ValueError: + # if something removed the key from the container + # when we read, ignore the ValueError that we would + # get otherwise. + pass + + self._append(key) + + return rv + + def __setitem__(self, key: t.Any, value: t.Any) -> None: + """Sets the value for an item. Moves the item up so that it + has the highest priority then. + """ + with self._wlock: + if key in self._mapping: + self._remove(key) + elif len(self._mapping) == self.capacity: + del self._mapping[self._popleft()] + + self._append(key) + self._mapping[key] = value + + def __delitem__(self, key: t.Any) -> None: + """Remove an item from the cache dict. + Raise a `KeyError` if it does not exist. + """ + with self._wlock: + del self._mapping[key] + + try: + self._remove(key) + except ValueError: + pass + + def items(self) -> t.Iterable[t.Tuple[t.Any, t.Any]]: + """Return a list of items.""" + result = [(key, self._mapping[key]) for key in list(self._queue)] + result.reverse() + return result + + def values(self) -> t.Iterable[t.Any]: + """Return a list of all values.""" + return [x[1] for x in self.items()] + + def keys(self) -> t.Iterable[t.Any]: + """Return a list of all keys ordered by most recent usage.""" + return list(self) + + def __iter__(self) -> t.Iterator[t.Any]: + return reversed(tuple(self._queue)) + + def __reversed__(self) -> t.Iterator[t.Any]: + """Iterate over the keys in the cache dict, oldest items + coming first. + """ + return iter(tuple(self._queue)) + + __copy__ = copy + + +def select_autoescape( + enabled_extensions: t.Collection[str] = ("html", "htm", "xml"), + disabled_extensions: t.Collection[str] = (), + default_for_string: bool = True, + default: bool = False, +) -> t.Callable[[t.Optional[str]], bool]: + """Intelligently sets the initial value of autoescaping based on the + filename of the template. This is the recommended way to configure + autoescaping if you do not want to write a custom function yourself. + + If you want to enable it for all templates created from strings or + for all templates with `.html` and `.xml` extensions:: + + from jinja2 import Environment, select_autoescape + env = Environment(autoescape=select_autoescape( + enabled_extensions=('html', 'xml'), + default_for_string=True, + )) + + Example configuration to turn it on at all times except if the template + ends with `.txt`:: + + from jinja2 import Environment, select_autoescape + env = Environment(autoescape=select_autoescape( + disabled_extensions=('txt',), + default_for_string=True, + default=True, + )) + + The `enabled_extensions` is an iterable of all the extensions that + autoescaping should be enabled for. Likewise `disabled_extensions` is + a list of all templates it should be disabled for. If a template is + loaded from a string then the default from `default_for_string` is used. + If nothing matches then the initial value of autoescaping is set to the + value of `default`. + + For security reasons this function operates case insensitive. + + .. versionadded:: 2.9 + """ + enabled_patterns = tuple(f".{x.lstrip('.').lower()}" for x in enabled_extensions) + disabled_patterns = tuple(f".{x.lstrip('.').lower()}" for x in disabled_extensions) + + def autoescape(template_name: t.Optional[str]) -> bool: + if template_name is None: + return default_for_string + template_name = template_name.lower() + if template_name.endswith(enabled_patterns): + return True + if template_name.endswith(disabled_patterns): + return False + return default + + return autoescape + + +def htmlsafe_json_dumps( + obj: t.Any, dumps: t.Optional[t.Callable[..., str]] = None, **kwargs: t.Any +) -> markupsafe.Markup: + """Serialize an object to a string of JSON with :func:`json.dumps`, + then replace HTML-unsafe characters with Unicode escapes and mark + the result safe with :class:`~markupsafe.Markup`. + + This is available in templates as the ``|tojson`` filter. + + The following characters are escaped: ``<``, ``>``, ``&``, ``'``. + + The returned string is safe to render in HTML documents and + ``