| """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, *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 # type: ignore |
| ): |
| __obj = __obj.__call__ # type: ignore |
| |
| 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 = concat( |
| [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 = concat(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 undefined type can be printed and |
| iterated over, but every other access 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: |
| if 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( # type: ignore |
| "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, _: str) -> "ChainableUndefined": |
| return self |
| |
| __getitem__ = __getattr__ # type: ignore |
| |
| |
| 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 |
| |
| |
| # Remove slots attributes, after the metaclass is applied they are |
| # unneeded and contain wrong data for subclasses. |
| del ( |
| Undefined.__slots__, |
| ChainableUndefined.__slots__, |
| DebugUndefined.__slots__, |
| StrictUndefined.__slots__, |
| ) |
