summaryrefslogtreecommitdiffhomepage
path: root/scripts/jinja2/environment.py
diff options
context:
space:
mode:
Diffstat (limited to 'scripts/jinja2/environment.py')
-rw-r--r--scripts/jinja2/environment.py1429
1 files changed, 942 insertions, 487 deletions
diff --git a/scripts/jinja2/environment.py b/scripts/jinja2/environment.py
index feb6df3..a231d9c 100644
--- a/scripts/jinja2/environment.py
+++ b/scripts/jinja2/environment.py
@@ -1,106 +1,146 @@
-# -*- coding: utf-8 -*-
-"""
- jinja2.environment
- ~~~~~~~~~~~~~~~~~~
-
- Provides a class that holds runtime and parsing time options.
-
- :copyright: (c) 2010 by the Jinja Team.
- :license: BSD, see LICENSE for more details.
+"""Classes for managing templates and their runtime and compile time
+options.
"""
import os
import sys
-from jinja2 import nodes
-from jinja2.defaults import BLOCK_START_STRING, \
- BLOCK_END_STRING, VARIABLE_START_STRING, VARIABLE_END_STRING, \
- COMMENT_START_STRING, COMMENT_END_STRING, LINE_STATEMENT_PREFIX, \
- LINE_COMMENT_PREFIX, TRIM_BLOCKS, NEWLINE_SEQUENCE, \
- DEFAULT_FILTERS, DEFAULT_TESTS, DEFAULT_NAMESPACE, \
- KEEP_TRAILING_NEWLINE, LSTRIP_BLOCKS
-from jinja2.lexer import get_lexer, TokenStream
-from jinja2.parser import Parser
-from jinja2.nodes import EvalContext
-from jinja2.optimizer import optimize
-from jinja2.compiler import generate
-from jinja2.runtime import Undefined, new_context
-from jinja2.exceptions import TemplateSyntaxError, TemplateNotFound, \
- TemplatesNotFound, TemplateRuntimeError
-from jinja2.utils import import_string, LRUCache, Markup, missing, \
- concat, consume, internalcode
-from jinja2._compat import imap, ifilter, string_types, iteritems, \
- text_type, reraise, implements_iterator, implements_to_string, \
- get_next, encode_filename, PY2, PYPY
+import typing
+import typing as t
+import weakref
+from collections import ChainMap
+from functools import lru_cache
+from functools import partial
from functools import reduce
+from types import CodeType
+
+from markupsafe import Markup
+
+from . import nodes
+from .compiler import CodeGenerator
+from .compiler import generate
+from .defaults import BLOCK_END_STRING
+from .defaults import BLOCK_START_STRING
+from .defaults import COMMENT_END_STRING
+from .defaults import COMMENT_START_STRING
+from .defaults import DEFAULT_FILTERS
+from .defaults import DEFAULT_NAMESPACE
+from .defaults import DEFAULT_POLICIES
+from .defaults import DEFAULT_TESTS
+from .defaults import KEEP_TRAILING_NEWLINE
+from .defaults import LINE_COMMENT_PREFIX
+from .defaults import LINE_STATEMENT_PREFIX
+from .defaults import LSTRIP_BLOCKS
+from .defaults import NEWLINE_SEQUENCE
+from .defaults import TRIM_BLOCKS
+from .defaults import VARIABLE_END_STRING
+from .defaults import VARIABLE_START_STRING
+from .exceptions import TemplateNotFound
+from .exceptions import TemplateRuntimeError
+from .exceptions import TemplatesNotFound
+from .exceptions import TemplateSyntaxError
+from .exceptions import UndefinedError
+from .lexer import get_lexer
+from .lexer import Lexer
+from .lexer import TokenStream
+from .nodes import EvalContext
+from .parser import Parser
+from .runtime import Context
+from .runtime import new_context
+from .runtime import Undefined
+from .utils import _PassArg
+from .utils import concat
+from .utils import consume
+from .utils import import_string
+from .utils import internalcode
+from .utils import LRUCache
+from .utils import missing
+
+if t.TYPE_CHECKING:
+ import typing_extensions as te
+ from .bccache import BytecodeCache
+ from .ext import Extension
+ from .loaders import BaseLoader
+
+_env_bound = t.TypeVar("_env_bound", bound="Environment")
# for direct template usage we have up to ten living environments
-_spontaneous_environments = LRUCache(10)
-
-# the function to create jinja traceback objects. This is dynamically
-# imported on the first exception in the exception handler.
-_make_traceback = None
-
-
-def get_spontaneous_environment(*args):
- """Return a new spontaneous environment. A spontaneous environment is an
- unnamed and unaccessible (in theory) environment that is used for
- templates generated from a string and not from the file system.
+@lru_cache(maxsize=10)
+def get_spontaneous_environment(cls: t.Type[_env_bound], *args: t.Any) -> _env_bound:
+ """Return a new spontaneous environment. A spontaneous environment
+ is used for templates created directly rather than through an
+ existing environment.
+
+ :param cls: Environment class to create.
+ :param args: Positional arguments passed to environment.
"""
- try:
- env = _spontaneous_environments.get(args)
- except TypeError:
- return Environment(*args)
- if env is not None:
- return env
- _spontaneous_environments[args] = env = Environment(*args)
+ env = cls(*args)
env.shared = True
return env
-def create_cache(size):
+def create_cache(
+ size: int,
+) -> t.Optional[t.MutableMapping[t.Tuple[weakref.ref, str], "Template"]]:
"""Return the cache class for the given size."""
if size == 0:
return None
+
if size < 0:
return {}
- return LRUCache(size)
+ return LRUCache(size) # type: ignore
-def copy_cache(cache):
+
+def copy_cache(
+ cache: t.Optional[t.MutableMapping],
+) -> t.Optional[t.MutableMapping[t.Tuple[weakref.ref, str], "Template"]]:
"""Create an empty copy of the given cache."""
if cache is None:
return None
- elif type(cache) is dict:
+
+ if type(cache) is dict:
return {}
- return LRUCache(cache.capacity)
+ return LRUCache(cache.capacity) # type: ignore
-def load_extensions(environment, extensions):
+
+def load_extensions(
+ environment: "Environment",
+ extensions: t.Sequence[t.Union[str, t.Type["Extension"]]],
+) -> t.Dict[str, "Extension"]:
"""Load the extensions from the list and bind it to the environment.
- Returns a dict of instantiated environments.
+ Returns a dict of instantiated extensions.
"""
result = {}
+
for extension in extensions:
- if isinstance(extension, string_types):
- extension = import_string(extension)
+ if isinstance(extension, str):
+ extension = t.cast(t.Type["Extension"], import_string(extension))
+
result[extension.identifier] = extension(environment)
+
return result
-def _environment_sanity_check(environment):
+def _environment_config_check(environment: "Environment") -> "Environment":
"""Perform a sanity check on the environment."""
- assert issubclass(environment.undefined, Undefined), 'undefined must ' \
- 'be a subclass of undefined because filters depend on it.'
- assert environment.block_start_string != \
- environment.variable_start_string != \
- environment.comment_start_string, 'block, variable and comment ' \
- 'start strings must be different'
- assert environment.newline_sequence in ('\r', '\r\n', '\n'), \
- 'newline_sequence set to unknown line ending string.'
+ assert issubclass(
+ environment.undefined, Undefined
+ ), "'undefined' must be a subclass of 'jinja2.Undefined'."
+ assert (
+ environment.block_start_string
+ != environment.variable_start_string
+ != environment.comment_start_string
+ ), "block, variable and comment start strings must be different."
+ assert environment.newline_sequence in {
+ "\r",
+ "\r\n",
+ "\n",
+ }, "'newline_sequence' must be one of '\\n', '\\r\\n', or '\\r'."
return environment
-class Environment(object):
+class Environment:
r"""The core component of Jinja is the `Environment`. It contains
important shared variables like configuration, filters, tests,
globals and others. Instances of this class may be modified if
@@ -136,7 +176,7 @@ class Environment(object):
`line_comment_prefix`
If given and a string, this will be used as prefix for line based
- based comments. See also :ref:`line-statements`.
+ comments. See also :ref:`line-statements`.
.. versionadded:: 2.2
@@ -167,7 +207,7 @@ class Environment(object):
look at :ref:`the extensions documentation <jinja-extensions>`.
`optimized`
- should the optimizer be enabled? Default is `True`.
+ should the optimizer be enabled? Default is ``True``.
`undefined`
:class:`Undefined` or a subclass of it that is used to represent
@@ -176,14 +216,14 @@ class Environment(object):
`finalize`
A callable that can be used to process the result of a variable
expression before it is output. For example one can convert
- `None` implicitly into an empty string here.
+ ``None`` implicitly into an empty string here.
`autoescape`
- If set to true the XML/HTML autoescaping feature is enabled by
+ If set to ``True`` the XML/HTML autoescaping feature is enabled by
default. For more details about autoescaping see
- :class:`~jinja2.utils.Markup`. As of Jinja 2.4 this can also
+ :class:`~markupsafe.Markup`. As of Jinja 2.4 this can also
be a callable that is passed the template name and has to
- return `True` or `False` depending on autoescape should be
+ return ``True`` or ``False`` depending on autoescape should be
enabled by default.
.. versionchanged:: 2.4
@@ -205,7 +245,7 @@ class Environment(object):
`auto_reload`
Some loaders load templates from locations where the template
sources may change (ie: file system or database). If
- `auto_reload` is set to `True` (default) every time a template is
+ ``auto_reload`` is set to ``True`` (default) every time a template is
requested the loader checks if the source changed and if yes, it
will reload the template. For higher performance it's possible to
disable that.
@@ -216,6 +256,10 @@ class Environment(object):
have to be parsed if they were not changed.
See :ref:`bytecode-cache` for more information.
+
+ `enable_async`
+ If set to true this enables async template execution which
+ allows using async functions and generators.
"""
#: if this environment is sandboxed. Modifying this variable won't make
@@ -228,38 +272,47 @@ class Environment(object):
overlayed = False
#: the environment this environment is linked to if it is an overlay
- linked_to = None
+ linked_to: t.Optional["Environment"] = None
#: shared environments have this set to `True`. A shared environment
#: must not be modified
shared = False
- #: these are currently EXPERIMENTAL undocumented features.
- exception_handler = None
- exception_formatter = None
-
- def __init__(self,
- block_start_string=BLOCK_START_STRING,
- block_end_string=BLOCK_END_STRING,
- variable_start_string=VARIABLE_START_STRING,
- variable_end_string=VARIABLE_END_STRING,
- comment_start_string=COMMENT_START_STRING,
- comment_end_string=COMMENT_END_STRING,
- line_statement_prefix=LINE_STATEMENT_PREFIX,
- line_comment_prefix=LINE_COMMENT_PREFIX,
- trim_blocks=TRIM_BLOCKS,
- lstrip_blocks=LSTRIP_BLOCKS,
- newline_sequence=NEWLINE_SEQUENCE,
- keep_trailing_newline=KEEP_TRAILING_NEWLINE,
- extensions=(),
- optimized=True,
- undefined=Undefined,
- finalize=None,
- autoescape=False,
- loader=None,
- cache_size=400,
- auto_reload=True,
- bytecode_cache=None):
+ #: the class that is used for code generation. See
+ #: :class:`~jinja2.compiler.CodeGenerator` for more information.
+ code_generator_class: t.Type["CodeGenerator"] = CodeGenerator
+
+ #: the context class that is used for templates. See
+ #: :class:`~jinja2.runtime.Context` for more information.
+ context_class: t.Type[Context] = Context
+
+ template_class: t.Type["Template"]
+
+ def __init__(
+ self,
+ block_start_string: str = BLOCK_START_STRING,
+ block_end_string: str = BLOCK_END_STRING,
+ variable_start_string: str = VARIABLE_START_STRING,
+ variable_end_string: str = VARIABLE_END_STRING,
+ comment_start_string: str = COMMENT_START_STRING,
+ comment_end_string: str = COMMENT_END_STRING,
+ line_statement_prefix: t.Optional[str] = LINE_STATEMENT_PREFIX,
+ line_comment_prefix: t.Optional[str] = LINE_COMMENT_PREFIX,
+ trim_blocks: bool = TRIM_BLOCKS,
+ lstrip_blocks: bool = LSTRIP_BLOCKS,
+ newline_sequence: "te.Literal['\\n', '\\r\\n', '\\r']" = NEWLINE_SEQUENCE,
+ keep_trailing_newline: bool = KEEP_TRAILING_NEWLINE,
+ extensions: t.Sequence[t.Union[str, t.Type["Extension"]]] = (),
+ optimized: bool = True,
+ undefined: t.Type[Undefined] = Undefined,
+ finalize: t.Optional[t.Callable[..., t.Any]] = None,
+ autoescape: t.Union[bool, t.Callable[[t.Optional[str]], bool]] = False,
+ loader: t.Optional["BaseLoader"] = None,
+ cache_size: int = 400,
+ auto_reload: bool = True,
+ bytecode_cache: t.Optional["BytecodeCache"] = None,
+ enable_async: bool = False,
+ ):
# !!Important notice!!
# The constructor accepts quite a few arguments that should be
# passed by keyword rather than position. However it's important to
@@ -286,7 +339,7 @@ class Environment(object):
self.keep_trailing_newline = keep_trailing_newline
# runtime information
- self.undefined = undefined
+ self.undefined: t.Type[Undefined] = undefined
self.optimized = optimized
self.finalize = finalize
self.autoescape = autoescape
@@ -302,36 +355,53 @@ class Environment(object):
self.bytecode_cache = bytecode_cache
self.auto_reload = auto_reload
+ # configurable policies
+ self.policies = DEFAULT_POLICIES.copy()
+
# load extensions
self.extensions = load_extensions(self, extensions)
- _environment_sanity_check(self)
+ self.is_async = enable_async
+ _environment_config_check(self)
- def add_extension(self, extension):
+ def add_extension(self, extension: t.Union[str, t.Type["Extension"]]) -> None:
"""Adds an extension after the environment was created.
.. versionadded:: 2.5
"""
self.extensions.update(load_extensions(self, [extension]))
- def extend(self, **attributes):
+ def extend(self, **attributes: t.Any) -> None:
"""Add the items to the instance of the environment if they do not exist
yet. This is used by :ref:`extensions <writing-extensions>` to register
callbacks and configuration values without breaking inheritance.
"""
- for key, value in iteritems(attributes):
+ for key, value in attributes.items():
if not hasattr(self, key):
setattr(self, key, value)
- def overlay(self, block_start_string=missing, block_end_string=missing,
- variable_start_string=missing, variable_end_string=missing,
- comment_start_string=missing, comment_end_string=missing,
- line_statement_prefix=missing, line_comment_prefix=missing,
- trim_blocks=missing, lstrip_blocks=missing,
- extensions=missing, optimized=missing,
- undefined=missing, finalize=missing, autoescape=missing,
- loader=missing, cache_size=missing, auto_reload=missing,
- bytecode_cache=missing):
+ def overlay(
+ self,
+ block_start_string: str = missing,
+ block_end_string: str = missing,
+ variable_start_string: str = missing,
+ variable_end_string: str = missing,
+ comment_start_string: str = missing,
+ comment_end_string: str = missing,
+ line_statement_prefix: t.Optional[str] = missing,
+ line_comment_prefix: t.Optional[str] = missing,
+ trim_blocks: bool = missing,
+ lstrip_blocks: bool = missing,
+ extensions: t.Sequence[t.Union[str, t.Type["Extension"]]] = missing,
+ optimized: bool = missing,
+ undefined: t.Type[Undefined] = missing,
+ finalize: t.Optional[t.Callable[..., t.Any]] = missing,
+ autoescape: t.Union[bool, t.Callable[[t.Optional[str]], bool]] = missing,
+ loader: t.Optional["BaseLoader"] = missing,
+ cache_size: int = missing,
+ auto_reload: bool = missing,
+ bytecode_cache: t.Optional["BytecodeCache"] = missing,
+ ) -> "Environment":
"""Create a new overlay environment that shares all the data with the
current environment except for cache and the overridden attributes.
Extensions cannot be removed for an overlayed environment. An overlayed
@@ -344,14 +414,14 @@ class Environment(object):
through.
"""
args = dict(locals())
- del args['self'], args['cache_size'], args['extensions']
+ del args["self"], args["cache_size"], args["extensions"]
rv = object.__new__(self.__class__)
rv.__dict__.update(self.__dict__)
rv.overlayed = True
rv.linked_to = self
- for key, value in iteritems(args):
+ for key, value in args.items():
if value is not missing:
setattr(rv, key, value)
@@ -361,26 +431,30 @@ class Environment(object):
rv.cache = copy_cache(self.cache)
rv.extensions = {}
- for key, value in iteritems(self.extensions):
+ for key, value in self.extensions.items():
rv.extensions[key] = value.bind(rv)
if extensions is not missing:
rv.extensions.update(load_extensions(rv, extensions))
- return _environment_sanity_check(rv)
+ return _environment_config_check(rv)
- lexer = property(get_lexer, doc="The lexer for this environment.")
+ @property
+ def lexer(self) -> Lexer:
+ """The lexer for this environment."""
+ return get_lexer(self)
- def iter_extensions(self):
+ def iter_extensions(self) -> t.Iterator["Extension"]:
"""Iterates over the extensions by priority."""
- return iter(sorted(self.extensions.values(),
- key=lambda x: x.priority))
+ return iter(sorted(self.extensions.values(), key=lambda x: x.priority))
- def getitem(self, obj, argument):
+ def getitem(
+ self, obj: t.Any, argument: t.Union[str, t.Any]
+ ) -> t.Union[t.Any, Undefined]:
"""Get an item or attribute of an object but prefer the item."""
try:
return obj[argument]
- except (TypeError, LookupError):
- if isinstance(argument, string_types):
+ except (AttributeError, TypeError, LookupError):
+ if isinstance(argument, str):
try:
attr = str(argument)
except Exception:
@@ -392,9 +466,9 @@ class Environment(object):
pass
return self.undefined(obj=obj, name=argument)
- def getattr(self, obj, attribute):
+ def getattr(self, obj: t.Any, attribute: str) -> t.Any:
"""Get an item or attribute of an object but prefer the attribute.
- Unlike :meth:`getitem` the attribute *must* be a bytestring.
+ Unlike :meth:`getitem` the attribute *must* be a string.
"""
try:
return getattr(obj, attribute)
@@ -405,63 +479,138 @@ class Environment(object):
except (TypeError, LookupError, AttributeError):
return self.undefined(obj=obj, name=attribute)
- def call_filter(self, name, value, args=None, kwargs=None,
- context=None, eval_ctx=None):
- """Invokes a filter on a value the same way the compiler does it.
+ def _filter_test_common(
+ self,
+ name: t.Union[str, Undefined],
+ value: t.Any,
+ args: t.Optional[t.Sequence[t.Any]],
+ kwargs: t.Optional[t.Mapping[str, t.Any]],
+ context: t.Optional[Context],
+ eval_ctx: t.Optional[EvalContext],
+ is_filter: bool,
+ ) -> t.Any:
+ if is_filter:
+ env_map = self.filters
+ type_name = "filter"
+ else:
+ env_map = self.tests
+ type_name = "test"
+
+ func = env_map.get(name) # type: ignore
- .. versionadded:: 2.7
- """
- func = self.filters.get(name)
if func is None:
- raise TemplateRuntimeError('no filter named %r' % name)
- args = [value] + list(args or ())
- if getattr(func, 'contextfilter', False):
+ msg = f"No {type_name} named {name!r}."
+
+ if isinstance(name, Undefined):
+ try:
+ name._fail_with_undefined_error()
+ except Exception as e:
+ msg = f"{msg} ({e}; did you forget to quote the callable name?)"
+
+ raise TemplateRuntimeError(msg)
+
+ args = [value, *(args if args is not None else ())]
+ kwargs = kwargs if kwargs is not None else {}
+ pass_arg = _PassArg.from_obj(func)
+
+ if pass_arg is _PassArg.context:
if context is None:
- raise TemplateRuntimeError('Attempted to invoke context '
- 'filter without context')
+ raise TemplateRuntimeError(
+ f"Attempted to invoke a context {type_name} without context."
+ )
+
args.insert(0, context)
- elif getattr(func, 'evalcontextfilter', False):
+ elif pass_arg is _PassArg.eval_context:
if eval_ctx is None:
if context is not None:
eval_ctx = context.eval_ctx
else:
eval_ctx = EvalContext(self)
+
args.insert(0, eval_ctx)
- elif getattr(func, 'environmentfilter', False):
+ elif pass_arg is _PassArg.environment:
args.insert(0, self)
- return func(*args, **(kwargs or {}))
- def call_test(self, name, value, args=None, kwargs=None):
- """Invokes a test on a value the same way the compiler does it.
+ return func(*args, **kwargs)
+
+ def call_filter(
+ self,
+ name: str,
+ value: t.Any,
+ args: t.Optional[t.Sequence[t.Any]] = None,
+ kwargs: t.Optional[t.Mapping[str, t.Any]] = None,
+ context: t.Optional[Context] = None,
+ eval_ctx: t.Optional[EvalContext] = None,
+ ) -> t.Any:
+ """Invoke a filter on a value the same way the compiler does.
+
+ This might return a coroutine if the filter is running from an
+ environment in async mode and the filter supports async
+ execution. It's your responsibility to await this if needed.
.. versionadded:: 2.7
"""
- func = self.tests.get(name)
- if func is None:
- raise TemplateRuntimeError('no test named %r' % name)
- return func(value, *(args or ()), **(kwargs or {}))
+ return self._filter_test_common(
+ name, value, args, kwargs, context, eval_ctx, True
+ )
+
+ def call_test(
+ self,
+ name: str,
+ value: t.Any,
+ args: t.Optional[t.Sequence[t.Any]] = None,
+ kwargs: t.Optional[t.Mapping[str, t.Any]] = None,
+ context: t.Optional[Context] = None,
+ eval_ctx: t.Optional[EvalContext] = None,
+ ) -> t.Any:
+ """Invoke a test on a value the same way the compiler does.
+
+ This might return a coroutine if the test is running from an
+ environment in async mode and the test supports async execution.
+ It's your responsibility to await this if needed.
+
+ .. versionchanged:: 3.0
+ Tests support ``@pass_context``, etc. decorators. Added
+ the ``context`` and ``eval_ctx`` parameters.
+
+ .. versionadded:: 2.7
+ """
+ return self._filter_test_common(
+ name, value, args, kwargs, context, eval_ctx, False
+ )
@internalcode
- def parse(self, source, name=None, filename=None):
+ def parse(
+ self,
+ source: str,
+ name: t.Optional[str] = None,
+ filename: t.Optional[str] = None,
+ ) -> nodes.Template:
"""Parse the sourcecode and return the abstract syntax tree. This
tree of nodes is used by the compiler to convert the template into
executable source- or bytecode. This is useful for debugging or to
extract information from templates.
- If you are :ref:`developing Jinja2 extensions <writing-extensions>`
+ If you are :ref:`developing Jinja extensions <writing-extensions>`
this gives you a good overview of the node tree generated.
"""
try:
return self._parse(source, name, filename)
except TemplateSyntaxError:
- exc_info = sys.exc_info()
- self.handle_exception(exc_info, source_hint=source)
+ self.handle_exception(source=source)
- def _parse(self, source, name, filename):
+ def _parse(
+ self, source: str, name: t.Optional[str], filename: t.Optional[str]
+ ) -> nodes.Template:
"""Internal parsing function used by `parse` and `compile`."""
- return Parser(self, source, name, encode_filename(filename)).parse()
-
- def lex(self, source, name=None, filename=None):
+ return Parser(self, source, name, filename).parse()
+
+ def lex(
+ self,
+ source: str,
+ name: t.Optional[str] = None,
+ filename: t.Optional[str] = None,
+ ) -> t.Iterator[t.Tuple[int, str, str]]:
"""Lex the given sourcecode and return a generator that yields
tokens as tuples in the form ``(lineno, token_type, value)``.
This can be useful for :ref:`extension development <writing-extensions>`
@@ -471,52 +620,109 @@ class Environment(object):
of the extensions to be applied you have to filter source through
the :meth:`preprocess` method.
"""
- source = text_type(source)
+ source = str(source)
try:
return self.lexer.tokeniter(source, name, filename)
except TemplateSyntaxError:
- exc_info = sys.exc_info()
- self.handle_exception(exc_info, source_hint=source)
-
- def preprocess(self, source, name=None, filename=None):
+ self.handle_exception(source=source)
+
+ def preprocess(
+ self,
+ source: str,
+ name: t.Optional[str] = None,
+ filename: t.Optional[str] = None,
+ ) -> str:
"""Preprocesses the source with all extensions. This is automatically
called for all parsing and compiling methods but *not* for :meth:`lex`
because there you usually only want the actual source tokenized.
"""
- return reduce(lambda s, e: e.preprocess(s, name, filename),
- self.iter_extensions(), text_type(source))
-
- def _tokenize(self, source, name, filename=None, state=None):
+ return reduce(
+ lambda s, e: e.preprocess(s, name, filename),
+ self.iter_extensions(),
+ str(source),
+ )
+
+ def _tokenize(
+ self,
+ source: str,
+ name: t.Optional[str],
+ filename: t.Optional[str] = None,
+ state: t.Optional[str] = None,
+ ) -> TokenStream:
"""Called by the parser to do the preprocessing and filtering
for all the extensions. Returns a :class:`~jinja2.lexer.TokenStream`.
"""
source = self.preprocess(source, name, filename)
stream = self.lexer.tokenize(source, name, filename, state)
+
for ext in self.iter_extensions():
- stream = ext.filter_stream(stream)
+ stream = ext.filter_stream(stream) # type: ignore
+
if not isinstance(stream, TokenStream):
- stream = TokenStream(stream, name, filename)
+ stream = TokenStream(stream, name, filename) # type: ignore
+
return stream
- def _generate(self, source, name, filename, defer_init=False):
+ def _generate(
+ self,
+ source: nodes.Template,
+ name: t.Optional[str],
+ filename: t.Optional[str],
+ defer_init: bool = False,
+ ) -> str:
"""Internal hook that can be overridden to hook a different generate
method in.
.. versionadded:: 2.5
"""
- return generate(source, self, name, filename, defer_init=defer_init)
-
- def _compile(self, source, filename):
+ return generate( # type: ignore
+ source,
+ self,
+ name,
+ filename,
+ defer_init=defer_init,
+ optimized=self.optimized,
+ )
+
+ def _compile(self, source: str, filename: str) -> CodeType:
"""Internal hook that can be overridden to hook a different compile
method in.
.. versionadded:: 2.5
"""
- return compile(source, filename, 'exec')
+ return compile(source, filename, "exec") # type: ignore
+
+ @typing.overload
+ def compile( # type: ignore
+ self,
+ source: t.Union[str, nodes.Template],
+ name: t.Optional[str] = None,
+ filename: t.Optional[str] = None,
+ raw: "te.Literal[False]" = False,
+ defer_init: bool = False,
+ ) -> CodeType:
+ ...
+
+ @typing.overload
+ def compile(
+ self,
+ source: t.Union[str, nodes.Template],
+ name: t.Optional[str] = None,
+ filename: t.Optional[str] = None,
+ raw: "te.Literal[True]" = ...,
+ defer_init: bool = False,
+ ) -> str:
+ ...
@internalcode
- def compile(self, source, name=None, filename=None, raw=False,
- defer_init=False):
+ def compile(
+ self,
+ source: t.Union[str, nodes.Template],
+ name: t.Optional[str] = None,
+ filename: t.Optional[str] = None,
+ raw: bool = False,
+ defer_init: bool = False,
+ ) -> t.Union[str, CodeType]:
"""Compile a node or template source code. The `name` parameter is
the load name of the template after it was joined using
:meth:`join_path` if necessary, not the filename on the file system.
@@ -538,25 +744,21 @@ class Environment(object):
"""
source_hint = None
try:
- if isinstance(source, string_types):
+ if isinstance(source, str):
source_hint = source
source = self._parse(source, name, filename)
- if self.optimized:
- source = optimize(source, self)
- source = self._generate(source, name, filename,
- defer_init=defer_init)
+ source = self._generate(source, name, filename, defer_init=defer_init)
if raw:
return source
if filename is None:
- filename = '<template>'
- else:
- filename = encode_filename(filename)
+ filename = "<template>"
return self._compile(source, filename)
except TemplateSyntaxError:
- exc_info = sys.exc_info()
- self.handle_exception(exc_info, source_hint=source_hint)
+ self.handle_exception(source=source_hint)
- def compile_expression(self, source, undefined_to_none=True):
+ def compile_expression(
+ self, source: str, undefined_to_none: bool = True
+ ) -> "TemplateExpression":
"""A handy helper method that returns a callable that accepts keyword
arguments that appear as variables in the expression. If called it
returns the result of the expression.
@@ -584,26 +786,30 @@ class Environment(object):
.. versionadded:: 2.1
"""
- parser = Parser(self, source, state='variable')
- exc_info = None
+ parser = Parser(self, source, state="variable")
try:
expr = parser.parse_expression()
if not parser.stream.eos:
- raise TemplateSyntaxError('chunk after expression',
- parser.stream.current.lineno,
- None, None)
+ raise TemplateSyntaxError(
+ "chunk after expression", parser.stream.current.lineno, None, None
+ )
expr.set_environment(self)
except TemplateSyntaxError:
- exc_info = sys.exc_info()
- if exc_info is not None:
- self.handle_exception(exc_info, source_hint=source)
- body = [nodes.Assign(nodes.Name('result', 'store'), expr, lineno=1)]
+ self.handle_exception(source=source)
+
+ body = [nodes.Assign(nodes.Name("result", "store"), expr, lineno=1)]
template = self.from_string(nodes.Template(body, lineno=1))
return TemplateExpression(template, undefined_to_none)
- def compile_templates(self, target, extensions=None, filter_func=None,
- zip='deflated', log_function=None,
- ignore_errors=True, py_compile=False):
+ def compile_templates(
+ self,
+ target: t.Union[str, os.PathLike],
+ extensions: t.Optional[t.Collection[str]] = None,
+ filter_func: t.Optional[t.Callable[[str], bool]] = None,
+ zip: t.Optional[str] = "deflated",
+ log_function: t.Optional[t.Callable[[str], None]] = None,
+ ignore_errors: bool = True,
+ ) -> None:
"""Finds all the templates the loader can find, compiles them
and stores them in `target`. If `zip` is `None`, instead of in a
zipfile, the templates will be stored in a directory.
@@ -619,54 +825,38 @@ class Environment(object):
syntax errors to abort the compilation you can set `ignore_errors`
to `False` and you will get an exception on syntax errors.
- If `py_compile` is set to `True` .pyc files will be written to the
- target instead of standard .py files. This flag does not do anything
- on pypy and Python 3 where pyc files are not picked up by itself and
- don't give much benefit.
-
.. versionadded:: 2.4
"""
- from jinja2.loaders import ModuleLoader
+ from .loaders import ModuleLoader
if log_function is None:
- log_function = lambda x: None
- if py_compile:
- if not PY2 or PYPY:
- from warnings import warn
- warn(Warning('py_compile has no effect on pypy or Python 3'))
- py_compile = False
- else:
- import importlib.util
- import marshal
- py_header = importlib.util.MAGIC_NUMBER + \
- u'\xff\xff\xff\xff'.encode('iso-8859-15')
+ def log_function(x: str) -> None:
+ pass
- # Python 3.3 added a source filesize to the header
- if sys.version_info >= (3, 3):
- py_header += u'\x00\x00\x00\x00'.encode('iso-8859-15')
+ assert log_function is not None
+ assert self.loader is not None, "No loader configured."
- def write_file(filename, data, mode):
+ def write_file(filename: str, data: str) -> None:
if zip:
info = ZipInfo(filename)
info.external_attr = 0o755 << 16
zip_file.writestr(info, data)
else:
- f = open(os.path.join(target, filename), mode)
- try:
- f.write(data)
- finally:
- f.close()
+ with open(os.path.join(target, filename), "wb") as f:
+ f.write(data.encode("utf8"))
if zip is not None:
from zipfile import ZipFile, ZipInfo, ZIP_DEFLATED, ZIP_STORED
- zip_file = ZipFile(target, 'w', dict(deflated=ZIP_DEFLATED,
- stored=ZIP_STORED)[zip])
- log_function('Compiling into Zip archive "%s"' % target)
+
+ zip_file = ZipFile(
+ target, "w", dict(deflated=ZIP_DEFLATED, stored=ZIP_STORED)[zip]
+ )
+ log_function(f"Compiling into Zip archive {target!r}")
else:
if not os.path.isdir(target):
os.makedirs(target)
- log_function('Compiling into folder "%s"' % target)
+ log_function(f"Compiling into folder {target!r}")
try:
for name in self.list_templates(extensions, filter_func):
@@ -676,27 +866,24 @@ class Environment(object):
except TemplateSyntaxError as e:
if not ignore_errors:
raise
- log_function('Could not compile "%s": %s' % (name, e))
+ log_function(f'Could not compile "{name}": {e}')
continue
filename = ModuleLoader.get_module_filename(name)
- if py_compile:
- c = self._compile(code, encode_filename(filename))
- write_file(filename + 'c', py_header +
- marshal.dumps(c), 'wb')
- log_function('Byte-compiled "%s" as %s' %
- (name, filename + 'c'))
- else:
- write_file(filename, code, 'w')
- log_function('Compiled "%s" as %s' % (name, filename))
+ write_file(filename, code)
+ log_function(f'Compiled "{name}" as {filename}')
finally:
if zip:
zip_file.close()
- log_function('Finished compiling templates')
+ log_function("Finished compiling templates")
- def list_templates(self, extensions=None, filter_func=None):
+ def list_templates(
+ self,
+ extensions: t.Optional[t.Collection[str]] = None,
+ filter_func: t.Optional[t.Callable[[str], bool]] = None,
+ ) -> t.List[str]:
"""Returns a list of templates for this environment. This requires
that the loader supports the loader's
:meth:`~BaseLoader.list_templates` method.
@@ -712,40 +899,32 @@ class Environment(object):
.. versionadded:: 2.4
"""
- x = self.loader.list_templates()
+ assert self.loader is not None, "No loader configured."
+ names = self.loader.list_templates()
+
if extensions is not None:
if filter_func is not None:
- raise TypeError('either extensions or filter_func '
- 'can be passed, but not both')
- filter_func = lambda x: '.' in x and \
- x.rsplit('.', 1)[1] in extensions
+ raise TypeError(
+ "either extensions or filter_func can be passed, but not both"
+ )
+
+ def filter_func(x: str) -> bool:
+ return "." in x and x.rsplit(".", 1)[1] in extensions # type: ignore
+
if filter_func is not None:
- x = list(ifilter(filter_func, x))
- return x
+ names = [name for name in names if filter_func(name)]
- def handle_exception(self, exc_info=None, rendered=False, source_hint=None):
+ return names
+
+ def handle_exception(self, source: t.Optional[str] = None) -> "te.NoReturn":
"""Exception handling helper. This is used internally to either raise
rewritten exceptions or return a rendered traceback for the template.
"""
- global _make_traceback
- if exc_info is None:
- exc_info = sys.exc_info()
-
- # the debugging module is imported when it's used for the first time.
- # we're doing a lot of stuff there and for applications that do not
- # get any exceptions in template rendering there is no need to load
- # all of that.
- if _make_traceback is None:
- from jinja2.debug import make_traceback as _make_traceback
- traceback = _make_traceback(exc_info, source_hint)
- if rendered and self.exception_formatter is not None:
- return self.exception_formatter(traceback)
- if self.exception_handler is not None:
- self.exception_handler(traceback)
- exc_type, exc_value, tb = traceback.standard_exc_info
- reraise(exc_type, exc_value, tb)
-
- def join_path(self, template, parent):
+ from .debug import rewrite_traceback_stack
+
+ raise rewrite_traceback_stack(source=source)
+
+ def join_path(self, template: str, parent: str) -> str:
"""Join a template with the parent. By default all the lookups are
relative to the loader root so this method returns the `template`
parameter unchanged, but if the paths should be relative to the
@@ -758,67 +937,108 @@ class Environment(object):
return template
@internalcode
- def _load_template(self, name, globals):
+ def _load_template(
+ self, name: str, globals: t.Optional[t.Mapping[str, t.Any]]
+ ) -> "Template":
if self.loader is None:
- raise TypeError('no loader for this environment specified')
- try:
- # use abs path for cache key
- cache_key = self.loader.get_source(self, name)[1]
- except RuntimeError:
- # if loader does not implement get_source()
- cache_key = None
- # if template is not file, use name for cache key
- if cache_key is None:
- cache_key = name
+ raise TypeError("no loader for this environment specified")
+ cache_key = (weakref.ref(self.loader), name)
if self.cache is not None:
template = self.cache.get(cache_key)
- if template is not None and (not self.auto_reload or
- template.is_up_to_date):
+ if template is not None and (
+ not self.auto_reload or template.is_up_to_date
+ ):
+ # template.globals is a ChainMap, modifying it will only
+ # affect the template, not the environment globals.
+ if globals:
+ template.globals.update(globals)
+
return template
- template = self.loader.load(self, name, globals)
+
+ template = self.loader.load(self, name, self.make_globals(globals))
+
if self.cache is not None:
self.cache[cache_key] = template
return template
@internalcode
- def get_template(self, name, parent=None, globals=None):
- """Load a template from the loader. If a loader is configured this
- method ask the loader for the template and returns a :class:`Template`.
- If the `parent` parameter is not `None`, :meth:`join_path` is called
- to get the real template name before loading.
-
- The `globals` parameter can be used to provide template wide globals.
- These variables are available in the context at render time.
-
- If the template does not exist a :exc:`TemplateNotFound` exception is
- raised.
+ def get_template(
+ self,
+ name: t.Union[str, "Template"],
+ parent: t.Optional[str] = None,
+ globals: t.Optional[t.Mapping[str, t.Any]] = None,
+ ) -> "Template":
+ """Load a template by name with :attr:`loader` and return a
+ :class:`Template`. If the template does not exist a
+ :exc:`TemplateNotFound` exception is raised.
+
+ :param name: Name of the template to load.
+ :param parent: The name of the parent template importing this
+ template. :meth:`join_path` can be used to implement name
+ transformations with this.
+ :param globals: Extend the environment :attr:`globals` with
+ these extra variables available for all renders of this
+ template. If the template has already been loaded and
+ cached, its globals are updated with any new items.
+
+ .. versionchanged:: 3.0
+ If a template is loaded from cache, ``globals`` will update
+ the template's globals instead of ignoring the new values.
.. versionchanged:: 2.4
- If `name` is a :class:`Template` object it is returned from the
- function unchanged.
+ If ``name`` is a :class:`Template` object it is returned
+ unchanged.
"""
if isinstance(name, Template):
return name
if parent is not None:
name = self.join_path(name, parent)
- return self._load_template(name, self.make_globals(globals))
- @internalcode
- def select_template(self, names, parent=None, globals=None):
- """Works like :meth:`get_template` but tries a number of templates
- before it fails. If it cannot find any of the templates, it will
- raise a :exc:`TemplatesNotFound` exception.
+ return self._load_template(name, globals)
- .. versionadded:: 2.3
+ @internalcode
+ def select_template(
+ self,
+ names: t.Iterable[t.Union[str, "Template"]],
+ parent: t.Optional[str] = None,
+ globals: t.Optional[t.Mapping[str, t.Any]] = None,
+ ) -> "Template":
+ """Like :meth:`get_template`, but tries loading multiple names.
+ If none of the names can be loaded a :exc:`TemplatesNotFound`
+ exception is raised.
+
+ :param names: List of template names to try loading in order.
+ :param parent: The name of the parent template importing this
+ template. :meth:`join_path` can be used to implement name
+ transformations with this.
+ :param globals: Extend the environment :attr:`globals` with
+ these extra variables available for all renders of this
+ template. If the template has already been loaded and
+ cached, its globals are updated with any new items.
+
+ .. versionchanged:: 3.0
+ If a template is loaded from cache, ``globals`` will update
+ the template's globals instead of ignoring the new values.
+
+ .. versionchanged:: 2.11
+ If ``names`` is :class:`Undefined`, an :exc:`UndefinedError`
+ is raised instead. If no templates were found and ``names``
+ contains :class:`Undefined`, the message is more helpful.
.. versionchanged:: 2.4
- If `names` contains a :class:`Template` object it is returned
- from the function unchanged.
+ If ``names`` contains a :class:`Template` object it is
+ returned unchanged.
+
+ .. versionadded:: 2.3
"""
+ if isinstance(names, Undefined):
+ names._fail_with_undefined_error()
+
if not names:
- raise TemplatesNotFound(message=u'Tried to select from an empty list '
- u'of templates.')
- globals = self.make_globals(globals)
+ raise TemplatesNotFound(
+ message="Tried to select from an empty list of templates."
+ )
+
for name in names:
if isinstance(name, Template):
return name
@@ -826,114 +1046,178 @@ class Environment(object):
name = self.join_path(name, parent)
try:
return self._load_template(name, globals)
- except TemplateNotFound:
+ except (TemplateNotFound, UndefinedError):
pass
- raise TemplatesNotFound(names)
+ raise TemplatesNotFound(names) # type: ignore
@internalcode
- def get_or_select_template(self, template_name_or_list,
- parent=None, globals=None):
- """Does a typecheck and dispatches to :meth:`select_template`
- if an iterable of template names is given, otherwise to
- :meth:`get_template`.
+ def get_or_select_template(
+ self,
+ template_name_or_list: t.Union[
+ str, "Template", t.List[t.Union[str, "Template"]]
+ ],
+ parent: t.Optional[str] = None,
+ globals: t.Optional[t.Mapping[str, t.Any]] = None,
+ ) -> "Template":
+ """Use :meth:`select_template` if an iterable of template names
+ is given, or :meth:`get_template` if one name is given.
.. versionadded:: 2.3
"""
- if isinstance(template_name_or_list, string_types):
+ if isinstance(template_name_or_list, (str, Undefined)):
return self.get_template(template_name_or_list, parent, globals)
elif isinstance(template_name_or_list, Template):
return template_name_or_list
return self.select_template(template_name_or_list, parent, globals)
- def from_string(self, source, globals=None, template_class=None):
- """Load a template from a string. This parses the source given and
- returns a :class:`Template` object.
+ def from_string(
+ self,
+ source: t.Union[str, nodes.Template],
+ globals: t.Optional[t.Mapping[str, t.Any]] = None,
+ template_class: t.Optional[t.Type["Template"]] = None,
+ ) -> "Template":
+ """Load a template from a source string without using
+ :attr:`loader`.
+
+ :param source: Jinja source to compile into a template.
+ :param globals: Extend the environment :attr:`globals` with
+ these extra variables available for all renders of this
+ template. If the template has already been loaded and
+ cached, its globals are updated with any new items.
+ :param template_class: Return an instance of this
+ :class:`Template` class.
"""
- globals = self.make_globals(globals)
+ gs = self.make_globals(globals)
cls = template_class or self.template_class
- return cls.from_code(self, self.compile(source), globals, None)
-
- def make_globals(self, d):
- """Return a dict for the globals."""
- if not d:
- return self.globals
- return dict(self.globals, **d)
-
-
-class Template(object):
- """The central template object. This class represents a compiled template
- and is used to evaluate it.
-
- Normally the template object is generated from an :class:`Environment` but
- it also has a constructor that makes it possible to create a template
- instance directly using the constructor. It takes the same arguments as
- the environment constructor but it's not possible to specify a loader.
-
- Every template object has a few methods and members that are guaranteed
- to exist. However it's important that a template object should be
- considered immutable. Modifications on the object are not supported.
-
- Template objects created from the constructor rather than an environment
- do have an `environment` attribute that points to a temporary environment
- that is probably shared with other templates created with the constructor
- and compatible settings.
-
- >>> template = Template('Hello {{ name }}!')
- >>> template.render(name='John Doe')
- u'Hello John Doe!'
-
- >>> stream = template.stream(name='John Doe')
- >>> stream.next()
- u'Hello John Doe!'
- >>> stream.next()
- Traceback (most recent call last):
- ...
- StopIteration
+ return cls.from_code(self, self.compile(source), gs, None)
+
+ def make_globals(
+ self, d: t.Optional[t.Mapping[str, t.Any]]
+ ) -> t.MutableMapping[str, t.Any]:
+ """Make the globals map for a template. Any given template
+ globals overlay the environment :attr:`globals`.
+
+ Returns a :class:`collections.ChainMap`. This allows any changes
+ to a template's globals to only affect that template, while
+ changes to the environment's globals are still reflected.
+ However, avoid modifying any globals after a template is loaded.
+
+ :param d: Dict of template-specific globals.
+
+ .. versionchanged:: 3.0
+ Use :class:`collections.ChainMap` to always prevent mutating
+ environment globals.
+ """
+ if d is None:
+ d = {}
+
+ return ChainMap(d, self.globals)
+
+
+class Template:
+ """A compiled template that can be rendered.
+
+ Use the methods on :class:`Environment` to create or load templates.
+ The environment is used to configure how templates are compiled and
+ behave.
+
+ It is also possible to create a template object directly. This is
+ not usually recommended. The constructor takes most of the same
+ arguments as :class:`Environment`. All templates created with the
+ same environment arguments share the same ephemeral ``Environment``
+ instance behind the scenes.
+
+ A template object should be considered immutable. Modifications on
+ the object are not supported.
"""
- def __new__(cls, source,
- block_start_string=BLOCK_START_STRING,
- block_end_string=BLOCK_END_STRING,
- variable_start_string=VARIABLE_START_STRING,
- variable_end_string=VARIABLE_END_STRING,
- comment_start_string=COMMENT_START_STRING,
- comment_end_string=COMMENT_END_STRING,
- line_statement_prefix=LINE_STATEMENT_PREFIX,
- line_comment_prefix=LINE_COMMENT_PREFIX,
- trim_blocks=TRIM_BLOCKS,
- lstrip_blocks=LSTRIP_BLOCKS,
- newline_sequence=NEWLINE_SEQUENCE,
- keep_trailing_newline=KEEP_TRAILING_NEWLINE,
- extensions=(),
- optimized=True,
- undefined=Undefined,
- finalize=None,
- autoescape=False):
+ #: Type of environment to create when creating a template directly
+ #: rather than through an existing environment.
+ environment_class: t.Type[Environment] = Environment
+
+ environment: Environment
+ globals: t.MutableMapping[str, t.Any]
+ name: t.Optional[str]
+ filename: t.Optional[str]
+ blocks: t.Dict[str, t.Callable[[Context], t.Iterator[str]]]
+ root_render_func: t.Callable[[Context], t.Iterator[str]]
+ _module: t.Optional["TemplateModule"]
+ _debug_info: str
+ _uptodate: t.Optional[t.Callable[[], bool]]
+
+ def __new__(
+ cls,
+ source: t.Union[str, nodes.Template],
+ block_start_string: str = BLOCK_START_STRING,
+ block_end_string: str = BLOCK_END_STRING,
+ variable_start_string: str = VARIABLE_START_STRING,
+ variable_end_string: str = VARIABLE_END_STRING,
+ comment_start_string: str = COMMENT_START_STRING,
+ comment_end_string: str = COMMENT_END_STRING,
+ line_statement_prefix: t.Optional[str] = LINE_STATEMENT_PREFIX,
+ line_comment_prefix: t.Optional[str] = LINE_COMMENT_PREFIX,
+ trim_blocks: bool = TRIM_BLOCKS,
+ lstrip_blocks: bool = LSTRIP_BLOCKS,
+ newline_sequence: "te.Literal['\\n', '\\r\\n', '\\r']" = NEWLINE_SEQUENCE,
+ keep_trailing_newline: bool = KEEP_TRAILING_NEWLINE,
+ extensions: t.Sequence[t.Union[str, t.Type["Extension"]]] = (),
+ optimized: bool = True,
+ undefined: t.Type[Undefined] = Undefined,
+ finalize: t.Optional[t.Callable[..., t.Any]] = None,
+ autoescape: t.Union[bool, t.Callable[[t.Optional[str]], bool]] = False,
+ enable_async: bool = False,
+ ) -> t.Any: # it returns a `Template`, but this breaks the sphinx build...
env = get_spontaneous_environment(
- block_start_string, block_end_string, variable_start_string,
- variable_end_string, comment_start_string, comment_end_string,
- line_statement_prefix, line_comment_prefix, trim_blocks,
- lstrip_blocks, newline_sequence, keep_trailing_newline,
- frozenset(extensions), optimized, undefined, finalize, autoescape,
- None, 0, False, None)
+ cls.environment_class, # type: ignore
+ block_start_string,
+ block_end_string,
+ variable_start_string,
+ variable_end_string,
+ comment_start_string,
+ comment_end_string,
+ line_statement_prefix,
+ line_comment_prefix,
+ trim_blocks,
+ lstrip_blocks,
+ newline_sequence,
+ keep_trailing_newline,
+ frozenset(extensions),
+ optimized,
+ undefined, # type: ignore
+ finalize,
+ autoescape,
+ None,
+ 0,
+ False,
+ None,
+ enable_async,
+ )
return env.from_string(source, template_class=cls)
@classmethod
- def from_code(cls, environment, code, globals, uptodate=None):
+ def from_code(
+ cls,
+ environment: Environment,
+ code: CodeType,
+ globals: t.MutableMapping[str, t.Any],
+ uptodate: t.Optional[t.Callable[[], bool]] = None,
+ ) -> "Template":
"""Creates a template object from compiled code and the globals. This
is used by the loaders and environment to create a template object.
"""
- namespace = {
- 'environment': environment,
- '__file__': code.co_filename
- }
+ namespace = {"environment": environment, "__file__": code.co_filename}
exec(code, namespace)
rv = cls._from_namespace(environment, namespace, globals)
rv._uptodate = uptodate
return rv
@classmethod
- def from_module_dict(cls, environment, module_dict, globals):
+ def from_module_dict(
+ cls,
+ environment: Environment,
+ module_dict: t.MutableMapping[str, t.Any],
+ globals: t.MutableMapping[str, t.Any],
+ ) -> "Template":
"""Creates a template object from a module. This is used by the
module loader to create a template object.
@@ -942,29 +1226,34 @@ class Template(object):
return cls._from_namespace(environment, module_dict, globals)
@classmethod
- def _from_namespace(cls, environment, namespace, globals):
- t = object.__new__(cls)
+ def _from_namespace(
+ cls,
+ environment: Environment,
+ namespace: t.MutableMapping[str, t.Any],
+ globals: t.MutableMapping[str, t.Any],
+ ) -> "Template":
+ t: "Template" = object.__new__(cls)
t.environment = environment
t.globals = globals
- t.name = namespace['name']
- t.filename = namespace['__file__']
- t.blocks = namespace['blocks']
+ t.name = namespace["name"]
+ t.filename = namespace["__file__"]
+ t.blocks = namespace["blocks"]
# render function and module
- t.root_render_func = namespace['root']
+ t.root_render_func = namespace["root"] # type: ignore
t._module = None
# debug and loader helpers
- t._debug_info = namespace['debug_info']
+ t._debug_info = namespace["debug_info"]
t._uptodate = None
# store the reference
- namespace['environment'] = environment
- namespace['__jinja_template__'] = t
+ namespace["environment"] = environment
+ namespace["__jinja_template__"] = t
return t
- def render(self, *args, **kwargs):
+ def render(self, *args: t.Any, **kwargs: t.Any) -> str:
"""This method accepts the same arguments as the `dict` constructor:
A dict, a dict subclass or some keyword arguments. If no arguments
are given the context will be empty. These two calls do the same::
@@ -972,77 +1261,217 @@ class Template(object):
template.render(knights='that say nih')
template.render({'knights': 'that say nih'})
- This will return the rendered template as unicode string.
+ This will return the rendered template as a string.
"""
- vars = dict(*args, **kwargs)
+ if self.environment.is_async:
+ import asyncio
+
+ close = False
+
+ if sys.version_info < (3, 7):
+ loop = asyncio.get_event_loop()
+ else:
+ try:
+ loop = asyncio.get_running_loop()
+ except RuntimeError:
+ loop = asyncio.new_event_loop()
+ close = True
+
+ try:
+ return loop.run_until_complete(self.render_async(*args, **kwargs))
+ finally:
+ if close:
+ loop.close()
+
+ ctx = self.new_context(dict(*args, **kwargs))
+
try:
- return concat(self.root_render_func(self.new_context(vars)))
+ return concat(self.root_render_func(ctx)) # type: ignore
except Exception:
- exc_info = sys.exc_info()
- return self.environment.handle_exception(exc_info, True)
+ self.environment.handle_exception()
+
+ async def render_async(self, *args: t.Any, **kwargs: t.Any) -> str:
+ """This works similar to :meth:`render` but returns a coroutine
+ that when awaited returns the entire rendered template string. This
+ requires the async feature to be enabled.
- def stream(self, *args, **kwargs):
+ Example usage::
+
+ await template.render_async(knights='that say nih; asynchronously')
+ """
+ if not self.environment.is_async:
+ raise RuntimeError(
+ "The environment was not created with async mode enabled."
+ )
+
+ ctx = self.new_context(dict(*args, **kwargs))
+
+ try:
+ return concat([n async for n in self.root_render_func(ctx)]) # type: ignore
+ except Exception:
+ return self.environment.handle_exception()
+
+ def stream(self, *args: t.Any, **kwargs: t.Any) -> "TemplateStream":
"""Works exactly like :meth:`generate` but returns a
:class:`TemplateStream`.
"""
return TemplateStream(self.generate(*args, **kwargs))
- def generate(self, *args, **kwargs):
+ def generate(self, *args: t.Any, **kwargs: t.Any) -> t.Iterator[str]:
"""For very large templates it can be useful to not render the whole
template at once but evaluate each statement after another and yield
piece for piece. This method basically does exactly that and returns
- a generator that yields one item after another as unicode strings.
+ a generator that yields one item after another as strings.
It accepts the same arguments as :meth:`render`.
"""
- vars = dict(*args, **kwargs)
+ if self.environment.is_async:
+ import asyncio
+
+ async def to_list() -> t.List[str]:
+ return [x async for x in self.generate_async(*args, **kwargs)]
+
+ if sys.version_info < (3, 7):
+ loop = asyncio.get_event_loop()
+ out = loop.run_until_complete(to_list())
+ else:
+ out = asyncio.run(to_list())
+
+ yield from out
+ return
+
+ ctx = self.new_context(dict(*args, **kwargs))
+
+ try:
+ yield from self.root_render_func(ctx) # type: ignore
+ except Exception:
+ yield self.environment.handle_exception()
+
+ async def generate_async(
+ self, *args: t.Any, **kwargs: t.Any
+ ) -> t.AsyncIterator[str]:
+ """An async version of :meth:`generate`. Works very similarly but
+ returns an async iterator instead.
+ """
+ if not self.environment.is_async:
+ raise RuntimeError(
+ "The environment was not created with async mode enabled."
+ )
+
+ ctx = self.new_context(dict(*args, **kwargs))
+
try:
- for event in self.root_render_func(self.new_context(vars)):
+ async for event in self.root_render_func(ctx): # type: ignore
yield event
except Exception:
- exc_info = sys.exc_info()
- else:
- return
- yield self.environment.handle_exception(exc_info, True)
-
- def new_context(self, vars=None, shared=False, locals=None):
+ yield self.environment.handle_exception()
+
+ def new_context(
+ self,
+ vars: t.Optional[t.Dict[str, t.Any]] = None,
+ shared: bool = False,
+ locals: t.Optional[t.Mapping[str, t.Any]] = None,
+ ) -> Context:
"""Create a new :class:`Context` for this template. The vars
provided will be passed to the template. Per default the globals
are added to the context. If shared is set to `True` the data
- is passed as it to the context without adding the globals.
+ is passed as is to the context without adding the globals.
`locals` can be a dict of local variables for internal usage.
"""
- return new_context(self.environment, self.name, self.blocks,
- vars, shared, self.globals, locals)
-
- def make_module(self, vars=None, shared=False, locals=None):
+ return new_context(
+ self.environment, self.name, self.blocks, vars, shared, self.globals, locals
+ )
+
+ def make_module(
+ self,
+ vars: t.Optional[t.Dict[str, t.Any]] = None,
+ shared: bool = False,
+ locals: t.Optional[t.Mapping[str, t.Any]] = None,
+ ) -> "TemplateModule":
"""This method works like the :attr:`module` attribute when called
without arguments but it will evaluate the template on every call
rather than caching it. It's also possible to provide
a dict which is then used as context. The arguments are the same
as for the :meth:`new_context` method.
"""
- return TemplateModule(self, self.new_context(vars, shared, locals))
+ ctx = self.new_context(vars, shared, locals)
+ return TemplateModule(self, ctx)
+
+ async def make_module_async(
+ self,
+ vars: t.Optional[t.Dict[str, t.Any]] = None,
+ shared: bool = False,
+ locals: t.Optional[t.Mapping[str, t.Any]] = None,
+ ) -> "TemplateModule":
+ """As template module creation can invoke template code for
+ asynchronous executions this method must be used instead of the
+ normal :meth:`make_module` one. Likewise the module attribute
+ becomes unavailable in async mode.
+ """
+ ctx = self.new_context(vars, shared, locals)
+ return TemplateModule(
+ self, ctx, [x async for x in self.root_render_func(ctx)] # type: ignore
+ )
+
+ @internalcode
+ def _get_default_module(self, ctx: t.Optional[Context] = None) -> "TemplateModule":
+ """If a context is passed in, this means that the template was
+ imported. Imported templates have access to the current
+ template's globals by default, but they can only be accessed via
+ the context during runtime.
+
+ If there are new globals, we need to create a new module because
+ the cached module is already rendered and will not have access
+ to globals from the current context. This new module is not
+ cached because the template can be imported elsewhere, and it
+ should have access to only the current template's globals.
+ """
+ if self.environment.is_async:
+ raise RuntimeError("Module is not available in async mode.")
+
+ if ctx is not None:
+ keys = ctx.globals_keys - self.globals.keys()
+
+ if keys:
+ return self.make_module({k: ctx.parent[k] for k in keys})
+
+ if self._module is None:
+ self._module = self.make_module()
+
+ return self._module
+
+ async def _get_default_module_async(
+ self, ctx: t.Optional[Context] = None
+ ) -> "TemplateModule":
+ if ctx is not None:
+ keys = ctx.globals_keys - self.globals.keys()
+
+ if keys:
+ return await self.make_module_async({k: ctx.parent[k] for k in keys})
+
+ if self._module is None:
+ self._module = await self.make_module_async()
+
+ return self._module
@property
- def module(self):
+ def module(self) -> "TemplateModule":
"""The template as module. This is used for imports in the
template runtime but is also useful if one wants to access
exported template variables from the Python layer:
>>> t = Template('{% macro foo() %}42{% endmacro %}23')
- >>> unicode(t.module)
- u'23'
- >>> t.module.foo()
- u'42'
+ >>> str(t.module)
+ '23'
+ >>> t.module.foo() == u'42'
+ True
+
+ This attribute is not available if async mode is enabled.
"""
- if self._module is not None:
- return self._module
- self._module = rv = self.make_module()
- return rv
+ return self._get_default_module()
- def get_corresponding_lineno(self, lineno):
+ def get_corresponding_lineno(self, lineno: int) -> int:
"""Return the source line number of a line number in the
generated bytecode as they are not in sync.
"""
@@ -1052,90 +1481,113 @@ class Template(object):
return 1
@property
- def is_up_to_date(self):
+ def is_up_to_date(self) -> bool:
"""If this variable is `False` there is a newer version available."""
if self._uptodate is None:
return True
return self._uptodate()
@property
- def debug_info(self):
+ def debug_info(self) -> t.List[t.Tuple[int, int]]:
"""The debug info mapping."""
- return [tuple(imap(int, x.split('='))) for x in
- self._debug_info.split('&')]
+ if self._debug_info:
+ return [
+ tuple(map(int, x.split("="))) # type: ignore
+ for x in self._debug_info.split("&")
+ ]
- def __repr__(self):
+ return []
+
+ def __repr__(self) -> str:
if self.name is None:
- name = 'memory:%x' % id(self)
+ name = f"memory:{id(self):x}"
else:
name = repr(self.name)
- return '<%s %s>' % (self.__class__.__name__, name)
+ return f"<{type(self).__name__} {name}>"
-@implements_to_string
-class TemplateModule(object):
+class TemplateModule:
"""Represents an imported template. All the exported names of the
template are available as attributes on this object. Additionally
- converting it into an unicode- or bytestrings renders the contents.
+ converting it into a string renders the contents.
"""
- def __init__(self, template, context):
- self._body_stream = list(template.root_render_func(context))
+ def __init__(
+ self,
+ template: Template,
+ context: Context,
+ body_stream: t.Optional[t.Iterable[str]] = None,
+ ) -> None:
+ if body_stream is None:
+ if context.environment.is_async:
+ raise RuntimeError(
+ "Async mode requires a body stream to be passed to"
+ " a template module. Use the async methods of the"
+ " API you are using."
+ )
+
+ body_stream = list(template.root_render_func(context)) # type: ignore
+
+ self._body_stream = body_stream
self.__dict__.update(context.get_exported())
self.__name__ = template.name
- def __html__(self):
+ def __html__(self) -> Markup:
return Markup(concat(self._body_stream))
- def __str__(self):
+ def __str__(self) -> str:
return concat(self._body_stream)
- def __repr__(self):
+ def __repr__(self) -> str:
if self.__name__ is None:
- name = 'memory:%x' % id(self)
+ name = f"memory:{id(self):x}"
else:
name = repr(self.__name__)
- return '<%s %s>' % (self.__class__.__name__, name)
+ return f"<{type(self).__name__} {name}>"
-class TemplateExpression(object):
+class TemplateExpression:
"""The :meth:`jinja2.Environment.compile_expression` method returns an
instance of this object. It encapsulates the expression-like access
to the template with an expression it wraps.
"""
- def __init__(self, template, undefined_to_none):
+ def __init__(self, template: Template, undefined_to_none: bool) -> None:
self._template = template
self._undefined_to_none = undefined_to_none
- def __call__(self, *args, **kwargs):
+ def __call__(self, *args: t.Any, **kwargs: t.Any) -> t.Optional[t.Any]:
context = self._template.new_context(dict(*args, **kwargs))
- consume(self._template.root_render_func(context))
- rv = context.vars['result']
+ consume(self._template.root_render_func(context)) # type: ignore
+ rv = context.vars["result"]
if self._undefined_to_none and isinstance(rv, Undefined):
rv = None
return rv
-@implements_iterator
-class TemplateStream(object):
+class TemplateStream:
"""A template stream works pretty much like an ordinary python generator
but it can buffer multiple items to reduce the number of total iterations.
Per default the output is unbuffered which means that for every unbuffered
- instruction in the template one unicode string is yielded.
+ instruction in the template one string is yielded.
If buffering is enabled with a buffer size of 5, five items are combined
- into a new unicode string. This is mainly useful if you are streaming
+ into a new string. This is mainly useful if you are streaming
big templates to a client via WSGI which flushes after each iteration.
"""
- def __init__(self, gen):
+ def __init__(self, gen: t.Iterator[str]) -> None:
self._gen = gen
self.disable_buffering()
- def dump(self, fp, encoding=None, errors='strict'):
+ def dump(
+ self,
+ fp: t.Union[str, t.IO],
+ encoding: t.Optional[str] = None,
+ errors: t.Optional[str] = "strict",
+ ) -> None:
"""Dump the complete stream into a file or file-like object.
- Per default unicode strings are written, if you want to encode
+ Per default strings are written, if you want to encode
before writing specify an `encoding`.
Example usage::
@@ -1143,17 +1595,20 @@ class TemplateStream(object):
Template('Hello {{ name }}!').stream(name='foo').dump('hello.html')
"""
close = False
- if isinstance(fp, string_types):
+
+ if isinstance(fp, str):
if encoding is None:
- encoding = 'utf-8'
- fp = open(fp, 'wb')
+ encoding = "utf-8"
+
+ fp = open(fp, "wb")
close = True
try:
if encoding is not None:
- iterable = (x.encode(encoding, errors) for x in self)
+ iterable = (x.encode(encoding, errors) for x in self) # type: ignore
else:
- iterable = self
- if hasattr(fp, 'writelines'):
+ iterable = self # type: ignore
+
+ if hasattr(fp, "writelines"):
fp.writelines(iterable)
else:
for item in iterable:
@@ -1162,43 +1617,43 @@ class TemplateStream(object):
if close:
fp.close()
- def disable_buffering(self):
+ def disable_buffering(self) -> None:
"""Disable the output buffering."""
- self._next = get_next(self._gen)
+ self._next = partial(next, self._gen)
self.buffered = False
- def enable_buffering(self, size=5):
- """Enable buffering. Buffer `size` items before yielding them."""
- if size <= 1:
- raise ValueError('buffer size too small')
+ def _buffered_generator(self, size: int) -> t.Iterator[str]:
+ buf: t.List[str] = []
+ c_size = 0
+ push = buf.append
- def generator(next):
- buf = []
+ while True:
+ try:
+ while c_size < size:
+ c = next(self._gen)
+ push(c)
+ if c:
+ c_size += 1
+ except StopIteration:
+ if not c_size:
+ return
+ yield concat(buf)
+ del buf[:]
c_size = 0
- push = buf.append
- while 1:
- try:
- while c_size < size:
- c = next()
- push(c)
- if c:
- c_size += 1
- except StopIteration:
- if not c_size:
- return
- yield concat(buf)
- del buf[:]
- c_size = 0
+ def enable_buffering(self, size: int = 5) -> None:
+ """Enable buffering. Buffer `size` items before yielding them."""
+ if size <= 1:
+ raise ValueError("buffer size too small")
self.buffered = True
- self._next = get_next(generator(get_next(self._gen)))
+ self._next = partial(next, self._buffered_generator(size))
- def __iter__(self):
+ def __iter__(self) -> "TemplateStream":
return self
- def __next__(self):
- return self._next()
+ def __next__(self) -> str:
+ return self._next() # type: ignore
# hook in default template class. if anyone reads this comment: ignore that