Загрузить файлы в «venv/Lib/site-packages/jinja2»
This commit is contained in:
870
venv/Lib/site-packages/jinja2/ext.py
Normal file
870
venv/Lib/site-packages/jinja2/ext.py
Normal file
@@ -0,0 +1,870 @@
|
||||
"""Extension API for adding custom tags and behavior."""
|
||||
|
||||
import pprint
|
||||
import re
|
||||
import typing as t
|
||||
|
||||
from markupsafe import Markup
|
||||
|
||||
from . import defaults
|
||||
from . import nodes
|
||||
from .environment import Environment
|
||||
from .exceptions import TemplateAssertionError
|
||||
from .exceptions import TemplateSyntaxError
|
||||
from .runtime import concat # type: ignore
|
||||
from .runtime import Context
|
||||
from .runtime import Undefined
|
||||
from .utils import import_string
|
||||
from .utils import pass_context
|
||||
|
||||
if t.TYPE_CHECKING:
|
||||
import typing_extensions as te
|
||||
|
||||
from .lexer import Token
|
||||
from .lexer import TokenStream
|
||||
from .parser import Parser
|
||||
|
||||
class _TranslationsBasic(te.Protocol):
|
||||
def gettext(self, message: str) -> str: ...
|
||||
|
||||
def ngettext(self, singular: str, plural: str, n: int) -> str:
|
||||
pass
|
||||
|
||||
class _TranslationsContext(_TranslationsBasic):
|
||||
def pgettext(self, context: str, message: str) -> str: ...
|
||||
|
||||
def npgettext(
|
||||
self, context: str, singular: str, plural: str, n: int
|
||||
) -> str: ...
|
||||
|
||||
_SupportedTranslations = t.Union[_TranslationsBasic, _TranslationsContext]
|
||||
|
||||
|
||||
# I18N functions available in Jinja templates. If the I18N library
|
||||
# provides ugettext, it will be assigned to gettext.
|
||||
GETTEXT_FUNCTIONS: t.Tuple[str, ...] = (
|
||||
"_",
|
||||
"gettext",
|
||||
"ngettext",
|
||||
"pgettext",
|
||||
"npgettext",
|
||||
)
|
||||
_ws_re = re.compile(r"\s*\n\s*")
|
||||
|
||||
|
||||
class Extension:
|
||||
"""Extensions can be used to add extra functionality to the Jinja template
|
||||
system at the parser level. Custom extensions are bound to an environment
|
||||
but may not store environment specific data on `self`. The reason for
|
||||
this is that an extension can be bound to another environment (for
|
||||
overlays) by creating a copy and reassigning the `environment` attribute.
|
||||
|
||||
As extensions are created by the environment they cannot accept any
|
||||
arguments for configuration. One may want to work around that by using
|
||||
a factory function, but that is not possible as extensions are identified
|
||||
by their import name. The correct way to configure the extension is
|
||||
storing the configuration values on the environment. Because this way the
|
||||
environment ends up acting as central configuration storage the
|
||||
attributes may clash which is why extensions have to ensure that the names
|
||||
they choose for configuration are not too generic. ``prefix`` for example
|
||||
is a terrible name, ``fragment_cache_prefix`` on the other hand is a good
|
||||
name as includes the name of the extension (fragment cache).
|
||||
"""
|
||||
|
||||
identifier: t.ClassVar[str]
|
||||
|
||||
def __init_subclass__(cls) -> None:
|
||||
cls.identifier = f"{cls.__module__}.{cls.__name__}"
|
||||
|
||||
#: if this extension parses this is the list of tags it's listening to.
|
||||
tags: t.Set[str] = set()
|
||||
|
||||
#: the priority of that extension. This is especially useful for
|
||||
#: extensions that preprocess values. A lower value means higher
|
||||
#: priority.
|
||||
#:
|
||||
#: .. versionadded:: 2.4
|
||||
priority = 100
|
||||
|
||||
def __init__(self, environment: Environment) -> None:
|
||||
self.environment = environment
|
||||
|
||||
def bind(self, environment: Environment) -> "te.Self":
|
||||
"""Create a copy of this extension bound to another environment."""
|
||||
rv = object.__new__(self.__class__)
|
||||
rv.__dict__.update(self.__dict__)
|
||||
rv.environment = environment
|
||||
return rv
|
||||
|
||||
def preprocess(
|
||||
self, source: str, name: t.Optional[str], filename: t.Optional[str] = None
|
||||
) -> str:
|
||||
"""This method is called before the actual lexing and can be used to
|
||||
preprocess the source. The `filename` is optional. The return value
|
||||
must be the preprocessed source.
|
||||
"""
|
||||
return source
|
||||
|
||||
def filter_stream(
|
||||
self, stream: "TokenStream"
|
||||
) -> t.Union["TokenStream", t.Iterable["Token"]]:
|
||||
"""It's passed a :class:`~jinja2.lexer.TokenStream` that can be used
|
||||
to filter tokens returned. This method has to return an iterable of
|
||||
:class:`~jinja2.lexer.Token`\\s, but it doesn't have to return a
|
||||
:class:`~jinja2.lexer.TokenStream`.
|
||||
"""
|
||||
return stream
|
||||
|
||||
def parse(self, parser: "Parser") -> t.Union[nodes.Node, t.List[nodes.Node]]:
|
||||
"""If any of the :attr:`tags` matched this method is called with the
|
||||
parser as first argument. The token the parser stream is pointing at
|
||||
is the name token that matched. This method has to return one or a
|
||||
list of multiple nodes.
|
||||
"""
|
||||
raise NotImplementedError()
|
||||
|
||||
def attr(
|
||||
self, name: str, lineno: t.Optional[int] = None
|
||||
) -> nodes.ExtensionAttribute:
|
||||
"""Return an attribute node for the current extension. This is useful
|
||||
to pass constants on extensions to generated template code.
|
||||
|
||||
::
|
||||
|
||||
self.attr('_my_attribute', lineno=lineno)
|
||||
"""
|
||||
return nodes.ExtensionAttribute(self.identifier, name, lineno=lineno)
|
||||
|
||||
def call_method(
|
||||
self,
|
||||
name: str,
|
||||
args: t.Optional[t.List[nodes.Expr]] = None,
|
||||
kwargs: t.Optional[t.List[nodes.Keyword]] = None,
|
||||
dyn_args: t.Optional[nodes.Expr] = None,
|
||||
dyn_kwargs: t.Optional[nodes.Expr] = None,
|
||||
lineno: t.Optional[int] = None,
|
||||
) -> nodes.Call:
|
||||
"""Call a method of the extension. This is a shortcut for
|
||||
:meth:`attr` + :class:`jinja2.nodes.Call`.
|
||||
"""
|
||||
if args is None:
|
||||
args = []
|
||||
if kwargs is None:
|
||||
kwargs = []
|
||||
return nodes.Call(
|
||||
self.attr(name, lineno=lineno),
|
||||
args,
|
||||
kwargs,
|
||||
dyn_args,
|
||||
dyn_kwargs,
|
||||
lineno=lineno,
|
||||
)
|
||||
|
||||
|
||||
@pass_context
|
||||
def _gettext_alias(
|
||||
__context: Context, *args: t.Any, **kwargs: t.Any
|
||||
) -> t.Union[t.Any, Undefined]:
|
||||
return __context.call(__context.resolve("gettext"), *args, **kwargs)
|
||||
|
||||
|
||||
def _make_new_gettext(func: t.Callable[[str], str]) -> t.Callable[..., str]:
|
||||
@pass_context
|
||||
def gettext(__context: Context, __string: str, **variables: t.Any) -> str:
|
||||
rv = __context.call(func, __string)
|
||||
if __context.eval_ctx.autoescape:
|
||||
rv = Markup(rv)
|
||||
# Always treat as a format string, even if there are no
|
||||
# variables. This makes translation strings more consistent
|
||||
# and predictable. This requires escaping
|
||||
return rv % variables # type: ignore
|
||||
|
||||
return gettext
|
||||
|
||||
|
||||
def _make_new_ngettext(func: t.Callable[[str, str, int], str]) -> t.Callable[..., str]:
|
||||
@pass_context
|
||||
def ngettext(
|
||||
__context: Context,
|
||||
__singular: str,
|
||||
__plural: str,
|
||||
__num: int,
|
||||
**variables: t.Any,
|
||||
) -> str:
|
||||
variables.setdefault("num", __num)
|
||||
rv = __context.call(func, __singular, __plural, __num)
|
||||
if __context.eval_ctx.autoescape:
|
||||
rv = Markup(rv)
|
||||
# Always treat as a format string, see gettext comment above.
|
||||
return rv % variables # type: ignore
|
||||
|
||||
return ngettext
|
||||
|
||||
|
||||
def _make_new_pgettext(func: t.Callable[[str, str], str]) -> t.Callable[..., str]:
|
||||
@pass_context
|
||||
def pgettext(
|
||||
__context: Context, __string_ctx: str, __string: str, **variables: t.Any
|
||||
) -> str:
|
||||
variables.setdefault("context", __string_ctx)
|
||||
rv = __context.call(func, __string_ctx, __string)
|
||||
|
||||
if __context.eval_ctx.autoescape:
|
||||
rv = Markup(rv)
|
||||
|
||||
# Always treat as a format string, see gettext comment above.
|
||||
return rv % variables # type: ignore
|
||||
|
||||
return pgettext
|
||||
|
||||
|
||||
def _make_new_npgettext(
|
||||
func: t.Callable[[str, str, str, int], str],
|
||||
) -> t.Callable[..., str]:
|
||||
@pass_context
|
||||
def npgettext(
|
||||
__context: Context,
|
||||
__string_ctx: str,
|
||||
__singular: str,
|
||||
__plural: str,
|
||||
__num: int,
|
||||
**variables: t.Any,
|
||||
) -> str:
|
||||
variables.setdefault("context", __string_ctx)
|
||||
variables.setdefault("num", __num)
|
||||
rv = __context.call(func, __string_ctx, __singular, __plural, __num)
|
||||
|
||||
if __context.eval_ctx.autoescape:
|
||||
rv = Markup(rv)
|
||||
|
||||
# Always treat as a format string, see gettext comment above.
|
||||
return rv % variables # type: ignore
|
||||
|
||||
return npgettext
|
||||
|
||||
|
||||
class InternationalizationExtension(Extension):
|
||||
"""This extension adds gettext support to Jinja."""
|
||||
|
||||
tags = {"trans"}
|
||||
|
||||
# TODO: the i18n extension is currently reevaluating values in a few
|
||||
# situations. Take this example:
|
||||
# {% trans count=something() %}{{ count }} foo{% pluralize
|
||||
# %}{{ count }} fooss{% endtrans %}
|
||||
# something is called twice here. One time for the gettext value and
|
||||
# the other time for the n-parameter of the ngettext function.
|
||||
|
||||
def __init__(self, environment: Environment) -> None:
|
||||
super().__init__(environment)
|
||||
environment.globals["_"] = _gettext_alias
|
||||
environment.extend(
|
||||
install_gettext_translations=self._install,
|
||||
install_null_translations=self._install_null,
|
||||
install_gettext_callables=self._install_callables,
|
||||
uninstall_gettext_translations=self._uninstall,
|
||||
extract_translations=self._extract,
|
||||
newstyle_gettext=False,
|
||||
)
|
||||
|
||||
def _install(
|
||||
self, translations: "_SupportedTranslations", newstyle: t.Optional[bool] = None
|
||||
) -> None:
|
||||
# ugettext and ungettext are preferred in case the I18N library
|
||||
# is providing compatibility with older Python versions.
|
||||
gettext = getattr(translations, "ugettext", None)
|
||||
if gettext is None:
|
||||
gettext = translations.gettext
|
||||
ngettext = getattr(translations, "ungettext", None)
|
||||
if ngettext is None:
|
||||
ngettext = translations.ngettext
|
||||
|
||||
pgettext = getattr(translations, "pgettext", None)
|
||||
npgettext = getattr(translations, "npgettext", None)
|
||||
self._install_callables(
|
||||
gettext, ngettext, newstyle=newstyle, pgettext=pgettext, npgettext=npgettext
|
||||
)
|
||||
|
||||
def _install_null(self, newstyle: t.Optional[bool] = None) -> None:
|
||||
import gettext
|
||||
|
||||
translations = gettext.NullTranslations()
|
||||
|
||||
if hasattr(translations, "pgettext"):
|
||||
# Python < 3.8
|
||||
pgettext = translations.pgettext
|
||||
else:
|
||||
|
||||
def pgettext(c: str, s: str) -> str: # type: ignore[misc]
|
||||
return s
|
||||
|
||||
if hasattr(translations, "npgettext"):
|
||||
npgettext = translations.npgettext
|
||||
else:
|
||||
|
||||
def npgettext(c: str, s: str, p: str, n: int) -> str: # type: ignore[misc]
|
||||
return s if n == 1 else p
|
||||
|
||||
self._install_callables(
|
||||
gettext=translations.gettext,
|
||||
ngettext=translations.ngettext,
|
||||
newstyle=newstyle,
|
||||
pgettext=pgettext,
|
||||
npgettext=npgettext,
|
||||
)
|
||||
|
||||
def _install_callables(
|
||||
self,
|
||||
gettext: t.Callable[[str], str],
|
||||
ngettext: t.Callable[[str, str, int], str],
|
||||
newstyle: t.Optional[bool] = None,
|
||||
pgettext: t.Optional[t.Callable[[str, str], str]] = None,
|
||||
npgettext: t.Optional[t.Callable[[str, str, str, int], str]] = None,
|
||||
) -> None:
|
||||
if newstyle is not None:
|
||||
self.environment.newstyle_gettext = newstyle # type: ignore
|
||||
if self.environment.newstyle_gettext: # type: ignore
|
||||
gettext = _make_new_gettext(gettext)
|
||||
ngettext = _make_new_ngettext(ngettext)
|
||||
|
||||
if pgettext is not None:
|
||||
pgettext = _make_new_pgettext(pgettext)
|
||||
|
||||
if npgettext is not None:
|
||||
npgettext = _make_new_npgettext(npgettext)
|
||||
|
||||
self.environment.globals.update(
|
||||
gettext=gettext, ngettext=ngettext, pgettext=pgettext, npgettext=npgettext
|
||||
)
|
||||
|
||||
def _uninstall(self, translations: "_SupportedTranslations") -> None:
|
||||
for key in ("gettext", "ngettext", "pgettext", "npgettext"):
|
||||
self.environment.globals.pop(key, None)
|
||||
|
||||
def _extract(
|
||||
self,
|
||||
source: t.Union[str, nodes.Template],
|
||||
gettext_functions: t.Sequence[str] = GETTEXT_FUNCTIONS,
|
||||
) -> t.Iterator[
|
||||
t.Tuple[int, str, t.Union[t.Optional[str], t.Tuple[t.Optional[str], ...]]]
|
||||
]:
|
||||
if isinstance(source, str):
|
||||
source = self.environment.parse(source)
|
||||
return extract_from_ast(source, gettext_functions)
|
||||
|
||||
def parse(self, parser: "Parser") -> t.Union[nodes.Node, t.List[nodes.Node]]:
|
||||
"""Parse a translatable tag."""
|
||||
lineno = next(parser.stream).lineno
|
||||
|
||||
context = None
|
||||
context_token = parser.stream.next_if("string")
|
||||
|
||||
if context_token is not None:
|
||||
context = context_token.value
|
||||
|
||||
# find all the variables referenced. Additionally a variable can be
|
||||
# defined in the body of the trans block too, but this is checked at
|
||||
# a later state.
|
||||
plural_expr: t.Optional[nodes.Expr] = None
|
||||
plural_expr_assignment: t.Optional[nodes.Assign] = None
|
||||
num_called_num = False
|
||||
variables: t.Dict[str, nodes.Expr] = {}
|
||||
trimmed = None
|
||||
while parser.stream.current.type != "block_end":
|
||||
if variables:
|
||||
parser.stream.expect("comma")
|
||||
|
||||
# skip colon for python compatibility
|
||||
if parser.stream.skip_if("colon"):
|
||||
break
|
||||
|
||||
token = parser.stream.expect("name")
|
||||
if token.value in variables:
|
||||
parser.fail(
|
||||
f"translatable variable {token.value!r} defined twice.",
|
||||
token.lineno,
|
||||
exc=TemplateAssertionError,
|
||||
)
|
||||
|
||||
# expressions
|
||||
if parser.stream.current.type == "assign":
|
||||
next(parser.stream)
|
||||
variables[token.value] = var = parser.parse_expression()
|
||||
elif trimmed is None and token.value in ("trimmed", "notrimmed"):
|
||||
trimmed = token.value == "trimmed"
|
||||
continue
|
||||
else:
|
||||
variables[token.value] = var = nodes.Name(token.value, "load")
|
||||
|
||||
if plural_expr is None:
|
||||
if isinstance(var, nodes.Call):
|
||||
plural_expr = nodes.Name("_trans", "load")
|
||||
variables[token.value] = plural_expr
|
||||
plural_expr_assignment = nodes.Assign(
|
||||
nodes.Name("_trans", "store"), var
|
||||
)
|
||||
else:
|
||||
plural_expr = var
|
||||
num_called_num = token.value == "num"
|
||||
|
||||
parser.stream.expect("block_end")
|
||||
|
||||
plural = None
|
||||
have_plural = False
|
||||
referenced = set()
|
||||
|
||||
# now parse until endtrans or pluralize
|
||||
singular_names, singular = self._parse_block(parser, True)
|
||||
if singular_names:
|
||||
referenced.update(singular_names)
|
||||
if plural_expr is None:
|
||||
plural_expr = nodes.Name(singular_names[0], "load")
|
||||
num_called_num = singular_names[0] == "num"
|
||||
|
||||
# if we have a pluralize block, we parse that too
|
||||
if parser.stream.current.test("name:pluralize"):
|
||||
have_plural = True
|
||||
next(parser.stream)
|
||||
if parser.stream.current.type != "block_end":
|
||||
token = parser.stream.expect("name")
|
||||
if token.value not in variables:
|
||||
parser.fail(
|
||||
f"unknown variable {token.value!r} for pluralization",
|
||||
token.lineno,
|
||||
exc=TemplateAssertionError,
|
||||
)
|
||||
plural_expr = variables[token.value]
|
||||
num_called_num = token.value == "num"
|
||||
parser.stream.expect("block_end")
|
||||
plural_names, plural = self._parse_block(parser, False)
|
||||
next(parser.stream)
|
||||
referenced.update(plural_names)
|
||||
else:
|
||||
next(parser.stream)
|
||||
|
||||
# register free names as simple name expressions
|
||||
for name in referenced:
|
||||
if name not in variables:
|
||||
variables[name] = nodes.Name(name, "load")
|
||||
|
||||
if not have_plural:
|
||||
plural_expr = None
|
||||
elif plural_expr is None:
|
||||
parser.fail("pluralize without variables", lineno)
|
||||
|
||||
if trimmed is None:
|
||||
trimmed = self.environment.policies["ext.i18n.trimmed"]
|
||||
if trimmed:
|
||||
singular = self._trim_whitespace(singular)
|
||||
if plural:
|
||||
plural = self._trim_whitespace(plural)
|
||||
|
||||
node = self._make_node(
|
||||
singular,
|
||||
plural,
|
||||
context,
|
||||
variables,
|
||||
plural_expr,
|
||||
bool(referenced),
|
||||
num_called_num and have_plural,
|
||||
)
|
||||
node.set_lineno(lineno)
|
||||
if plural_expr_assignment is not None:
|
||||
return [plural_expr_assignment, node]
|
||||
else:
|
||||
return node
|
||||
|
||||
def _trim_whitespace(self, string: str, _ws_re: t.Pattern[str] = _ws_re) -> str:
|
||||
return _ws_re.sub(" ", string.strip())
|
||||
|
||||
def _parse_block(
|
||||
self, parser: "Parser", allow_pluralize: bool
|
||||
) -> t.Tuple[t.List[str], str]:
|
||||
"""Parse until the next block tag with a given name."""
|
||||
referenced = []
|
||||
buf = []
|
||||
|
||||
while True:
|
||||
if parser.stream.current.type == "data":
|
||||
buf.append(parser.stream.current.value.replace("%", "%%"))
|
||||
next(parser.stream)
|
||||
elif parser.stream.current.type == "variable_begin":
|
||||
next(parser.stream)
|
||||
name = parser.stream.expect("name").value
|
||||
referenced.append(name)
|
||||
buf.append(f"%({name})s")
|
||||
parser.stream.expect("variable_end")
|
||||
elif parser.stream.current.type == "block_begin":
|
||||
next(parser.stream)
|
||||
block_name = (
|
||||
parser.stream.current.value
|
||||
if parser.stream.current.type == "name"
|
||||
else None
|
||||
)
|
||||
if block_name == "endtrans":
|
||||
break
|
||||
elif block_name == "pluralize":
|
||||
if allow_pluralize:
|
||||
break
|
||||
parser.fail(
|
||||
"a translatable section can have only one pluralize section"
|
||||
)
|
||||
elif block_name == "trans":
|
||||
parser.fail(
|
||||
"trans blocks can't be nested; did you mean `endtrans`?"
|
||||
)
|
||||
parser.fail(
|
||||
f"control structures in translatable sections are not allowed; "
|
||||
f"saw `{block_name}`"
|
||||
)
|
||||
elif parser.stream.eos:
|
||||
parser.fail("unclosed translation block")
|
||||
else:
|
||||
raise RuntimeError("internal parser error")
|
||||
|
||||
return referenced, concat(buf)
|
||||
|
||||
def _make_node(
|
||||
self,
|
||||
singular: str,
|
||||
plural: t.Optional[str],
|
||||
context: t.Optional[str],
|
||||
variables: t.Dict[str, nodes.Expr],
|
||||
plural_expr: t.Optional[nodes.Expr],
|
||||
vars_referenced: bool,
|
||||
num_called_num: bool,
|
||||
) -> nodes.Output:
|
||||
"""Generates a useful node from the data provided."""
|
||||
newstyle = self.environment.newstyle_gettext # type: ignore
|
||||
node: nodes.Expr
|
||||
|
||||
# no variables referenced? no need to escape for old style
|
||||
# gettext invocations only if there are vars.
|
||||
if not vars_referenced and not newstyle:
|
||||
singular = singular.replace("%%", "%")
|
||||
if plural:
|
||||
plural = plural.replace("%%", "%")
|
||||
|
||||
func_name = "gettext"
|
||||
func_args: t.List[nodes.Expr] = [nodes.Const(singular)]
|
||||
|
||||
if context is not None:
|
||||
func_args.insert(0, nodes.Const(context))
|
||||
func_name = f"p{func_name}"
|
||||
|
||||
if plural_expr is not None:
|
||||
func_name = f"n{func_name}"
|
||||
func_args.extend((nodes.Const(plural), plural_expr))
|
||||
|
||||
node = nodes.Call(nodes.Name(func_name, "load"), func_args, [], None, None)
|
||||
|
||||
# in case newstyle gettext is used, the method is powerful
|
||||
# enough to handle the variable expansion and autoescape
|
||||
# handling itself
|
||||
if newstyle:
|
||||
for key, value in variables.items():
|
||||
# the function adds that later anyways in case num was
|
||||
# called num, so just skip it.
|
||||
if num_called_num and key == "num":
|
||||
continue
|
||||
node.kwargs.append(nodes.Keyword(key, value))
|
||||
|
||||
# otherwise do that here
|
||||
else:
|
||||
# mark the return value as safe if we are in an
|
||||
# environment with autoescaping turned on
|
||||
node = nodes.MarkSafeIfAutoescape(node)
|
||||
if variables:
|
||||
node = nodes.Mod(
|
||||
node,
|
||||
nodes.Dict(
|
||||
[
|
||||
nodes.Pair(nodes.Const(key), value)
|
||||
for key, value in variables.items()
|
||||
]
|
||||
),
|
||||
)
|
||||
return nodes.Output([node])
|
||||
|
||||
|
||||
class ExprStmtExtension(Extension):
|
||||
"""Adds a `do` tag to Jinja that works like the print statement just
|
||||
that it doesn't print the return value.
|
||||
"""
|
||||
|
||||
tags = {"do"}
|
||||
|
||||
def parse(self, parser: "Parser") -> nodes.ExprStmt:
|
||||
node = nodes.ExprStmt(lineno=next(parser.stream).lineno)
|
||||
node.node = parser.parse_tuple()
|
||||
return node
|
||||
|
||||
|
||||
class LoopControlExtension(Extension):
|
||||
"""Adds break and continue to the template engine."""
|
||||
|
||||
tags = {"break", "continue"}
|
||||
|
||||
def parse(self, parser: "Parser") -> t.Union[nodes.Break, nodes.Continue]:
|
||||
token = next(parser.stream)
|
||||
if token.value == "break":
|
||||
return nodes.Break(lineno=token.lineno)
|
||||
return nodes.Continue(lineno=token.lineno)
|
||||
|
||||
|
||||
class DebugExtension(Extension):
|
||||
"""A ``{% debug %}`` tag that dumps the available variables,
|
||||
filters, and tests.
|
||||
|
||||
.. code-block:: html+jinja
|
||||
|
||||
<pre>{% debug %}</pre>
|
||||
|
||||
.. code-block:: text
|
||||
|
||||
{'context': {'cycler': <class 'jinja2.utils.Cycler'>,
|
||||
...,
|
||||
'namespace': <class 'jinja2.utils.Namespace'>},
|
||||
'filters': ['abs', 'attr', 'batch', 'capitalize', 'center', 'count', 'd',
|
||||
..., 'urlencode', 'urlize', 'wordcount', 'wordwrap', 'xmlattr'],
|
||||
'tests': ['!=', '<', '<=', '==', '>', '>=', 'callable', 'defined',
|
||||
..., 'odd', 'sameas', 'sequence', 'string', 'undefined', 'upper']}
|
||||
|
||||
.. versionadded:: 2.11.0
|
||||
"""
|
||||
|
||||
tags = {"debug"}
|
||||
|
||||
def parse(self, parser: "Parser") -> nodes.Output:
|
||||
lineno = parser.stream.expect("name:debug").lineno
|
||||
context = nodes.ContextReference()
|
||||
result = self.call_method("_render", [context], lineno=lineno)
|
||||
return nodes.Output([result], lineno=lineno)
|
||||
|
||||
def _render(self, context: Context) -> str:
|
||||
result = {
|
||||
"context": context.get_all(),
|
||||
"filters": sorted(self.environment.filters.keys()),
|
||||
"tests": sorted(self.environment.tests.keys()),
|
||||
}
|
||||
|
||||
# Set the depth since the intent is to show the top few names.
|
||||
return pprint.pformat(result, depth=3, compact=True)
|
||||
|
||||
|
||||
def extract_from_ast(
|
||||
ast: nodes.Template,
|
||||
gettext_functions: t.Sequence[str] = GETTEXT_FUNCTIONS,
|
||||
babel_style: bool = True,
|
||||
) -> t.Iterator[
|
||||
t.Tuple[int, str, t.Union[t.Optional[str], t.Tuple[t.Optional[str], ...]]]
|
||||
]:
|
||||
"""Extract localizable strings from the given template node. Per
|
||||
default this function returns matches in babel style that means non string
|
||||
parameters as well as keyword arguments are returned as `None`. This
|
||||
allows Babel to figure out what you really meant if you are using
|
||||
gettext functions that allow keyword arguments for placeholder expansion.
|
||||
If you don't want that behavior set the `babel_style` parameter to `False`
|
||||
which causes only strings to be returned and parameters are always stored
|
||||
in tuples. As a consequence invalid gettext calls (calls without a single
|
||||
string parameter or string parameters after non-string parameters) are
|
||||
skipped.
|
||||
|
||||
This example explains the behavior:
|
||||
|
||||
>>> from jinja2 import Environment
|
||||
>>> env = Environment()
|
||||
>>> node = env.parse('{{ (_("foo"), _(), ngettext("foo", "bar", 42)) }}')
|
||||
>>> list(extract_from_ast(node))
|
||||
[(1, '_', 'foo'), (1, '_', ()), (1, 'ngettext', ('foo', 'bar', None))]
|
||||
>>> list(extract_from_ast(node, babel_style=False))
|
||||
[(1, '_', ('foo',)), (1, 'ngettext', ('foo', 'bar'))]
|
||||
|
||||
For every string found this function yields a ``(lineno, function,
|
||||
message)`` tuple, where:
|
||||
|
||||
* ``lineno`` is the number of the line on which the string was found,
|
||||
* ``function`` is the name of the ``gettext`` function used (if the
|
||||
string was extracted from embedded Python code), and
|
||||
* ``message`` is the string, or a tuple of strings for functions
|
||||
with multiple string arguments.
|
||||
|
||||
This extraction function operates on the AST and is because of that unable
|
||||
to extract any comments. For comment support you have to use the babel
|
||||
extraction interface or extract comments yourself.
|
||||
"""
|
||||
out: t.Union[t.Optional[str], t.Tuple[t.Optional[str], ...]]
|
||||
|
||||
for node in ast.find_all(nodes.Call):
|
||||
if (
|
||||
not isinstance(node.node, nodes.Name)
|
||||
or node.node.name not in gettext_functions
|
||||
):
|
||||
continue
|
||||
|
||||
strings: t.List[t.Optional[str]] = []
|
||||
|
||||
for arg in node.args:
|
||||
if isinstance(arg, nodes.Const) and isinstance(arg.value, str):
|
||||
strings.append(arg.value)
|
||||
else:
|
||||
strings.append(None)
|
||||
|
||||
for _ in node.kwargs:
|
||||
strings.append(None)
|
||||
if node.dyn_args is not None:
|
||||
strings.append(None)
|
||||
if node.dyn_kwargs is not None:
|
||||
strings.append(None)
|
||||
|
||||
if not babel_style:
|
||||
out = tuple(x for x in strings if x is not None)
|
||||
|
||||
if not out:
|
||||
continue
|
||||
else:
|
||||
if len(strings) == 1:
|
||||
out = strings[0]
|
||||
else:
|
||||
out = tuple(strings)
|
||||
|
||||
yield node.lineno, node.node.name, out
|
||||
|
||||
|
||||
class _CommentFinder:
|
||||
"""Helper class to find comments in a token stream. Can only
|
||||
find comments for gettext calls forwards. Once the comment
|
||||
from line 4 is found, a comment for line 1 will not return a
|
||||
usable value.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self, tokens: t.Sequence[t.Tuple[int, str, str]], comment_tags: t.Sequence[str]
|
||||
) -> None:
|
||||
self.tokens = tokens
|
||||
self.comment_tags = comment_tags
|
||||
self.offset = 0
|
||||
self.last_lineno = 0
|
||||
|
||||
def find_backwards(self, offset: int) -> t.List[str]:
|
||||
try:
|
||||
for _, token_type, token_value in reversed(
|
||||
self.tokens[self.offset : offset]
|
||||
):
|
||||
if token_type in ("comment", "linecomment"):
|
||||
try:
|
||||
prefix, comment = token_value.split(None, 1)
|
||||
except ValueError:
|
||||
continue
|
||||
if prefix in self.comment_tags:
|
||||
return [comment.rstrip()]
|
||||
return []
|
||||
finally:
|
||||
self.offset = offset
|
||||
|
||||
def find_comments(self, lineno: int) -> t.List[str]:
|
||||
if not self.comment_tags or self.last_lineno > lineno:
|
||||
return []
|
||||
for idx, (token_lineno, _, _) in enumerate(self.tokens[self.offset :]):
|
||||
if token_lineno > lineno:
|
||||
return self.find_backwards(self.offset + idx)
|
||||
return self.find_backwards(len(self.tokens))
|
||||
|
||||
|
||||
def babel_extract(
|
||||
fileobj: t.BinaryIO,
|
||||
keywords: t.Sequence[str],
|
||||
comment_tags: t.Sequence[str],
|
||||
options: t.Dict[str, t.Any],
|
||||
) -> t.Iterator[
|
||||
t.Tuple[
|
||||
int, str, t.Union[t.Optional[str], t.Tuple[t.Optional[str], ...]], t.List[str]
|
||||
]
|
||||
]:
|
||||
"""Babel extraction method for Jinja templates.
|
||||
|
||||
.. versionchanged:: 2.3
|
||||
Basic support for translation comments was added. If `comment_tags`
|
||||
is now set to a list of keywords for extraction, the extractor will
|
||||
try to find the best preceding comment that begins with one of the
|
||||
keywords. For best results, make sure to not have more than one
|
||||
gettext call in one line of code and the matching comment in the
|
||||
same line or the line before.
|
||||
|
||||
.. versionchanged:: 2.5.1
|
||||
The `newstyle_gettext` flag can be set to `True` to enable newstyle
|
||||
gettext calls.
|
||||
|
||||
.. versionchanged:: 2.7
|
||||
A `silent` option can now be provided. If set to `False` template
|
||||
syntax errors are propagated instead of being ignored.
|
||||
|
||||
:param fileobj: the file-like object the messages should be extracted from
|
||||
:param keywords: a list of keywords (i.e. function names) that should be
|
||||
recognized as translation functions
|
||||
:param comment_tags: a list of translator tags to search for and include
|
||||
in the results.
|
||||
:param options: a dictionary of additional options (optional)
|
||||
:return: an iterator over ``(lineno, funcname, message, comments)`` tuples.
|
||||
(comments will be empty currently)
|
||||
"""
|
||||
extensions: t.Dict[t.Type[Extension], None] = {}
|
||||
|
||||
for extension_name in options.get("extensions", "").split(","):
|
||||
extension_name = extension_name.strip()
|
||||
|
||||
if not extension_name:
|
||||
continue
|
||||
|
||||
extensions[import_string(extension_name)] = None
|
||||
|
||||
if InternationalizationExtension not in extensions:
|
||||
extensions[InternationalizationExtension] = None
|
||||
|
||||
def getbool(options: t.Mapping[str, str], key: str, default: bool = False) -> bool:
|
||||
return options.get(key, str(default)).lower() in {"1", "on", "yes", "true"}
|
||||
|
||||
silent = getbool(options, "silent", True)
|
||||
environment = Environment(
|
||||
options.get("block_start_string", defaults.BLOCK_START_STRING),
|
||||
options.get("block_end_string", defaults.BLOCK_END_STRING),
|
||||
options.get("variable_start_string", defaults.VARIABLE_START_STRING),
|
||||
options.get("variable_end_string", defaults.VARIABLE_END_STRING),
|
||||
options.get("comment_start_string", defaults.COMMENT_START_STRING),
|
||||
options.get("comment_end_string", defaults.COMMENT_END_STRING),
|
||||
options.get("line_statement_prefix") or defaults.LINE_STATEMENT_PREFIX,
|
||||
options.get("line_comment_prefix") or defaults.LINE_COMMENT_PREFIX,
|
||||
getbool(options, "trim_blocks", defaults.TRIM_BLOCKS),
|
||||
getbool(options, "lstrip_blocks", defaults.LSTRIP_BLOCKS),
|
||||
defaults.NEWLINE_SEQUENCE,
|
||||
getbool(options, "keep_trailing_newline", defaults.KEEP_TRAILING_NEWLINE),
|
||||
tuple(extensions),
|
||||
cache_size=0,
|
||||
auto_reload=False,
|
||||
)
|
||||
|
||||
if getbool(options, "trimmed"):
|
||||
environment.policies["ext.i18n.trimmed"] = True
|
||||
if getbool(options, "newstyle_gettext"):
|
||||
environment.newstyle_gettext = True # type: ignore
|
||||
|
||||
source = fileobj.read().decode(options.get("encoding", "utf-8"))
|
||||
try:
|
||||
node = environment.parse(source)
|
||||
tokens = list(environment.lex(environment.preprocess(source)))
|
||||
except TemplateSyntaxError:
|
||||
if not silent:
|
||||
raise
|
||||
# skip templates with syntax errors
|
||||
return
|
||||
|
||||
finder = _CommentFinder(tokens, comment_tags)
|
||||
for lineno, func, message in extract_from_ast(node, keywords):
|
||||
yield lineno, func, message, finder.find_comments(lineno)
|
||||
|
||||
|
||||
#: nicer import names
|
||||
i18n = InternationalizationExtension
|
||||
do = ExprStmtExtension
|
||||
loopcontrols = LoopControlExtension
|
||||
debug = DebugExtension
|
||||
1873
venv/Lib/site-packages/jinja2/filters.py
Normal file
1873
venv/Lib/site-packages/jinja2/filters.py
Normal file
File diff suppressed because it is too large
Load Diff
318
venv/Lib/site-packages/jinja2/idtracking.py
Normal file
318
venv/Lib/site-packages/jinja2/idtracking.py
Normal file
@@ -0,0 +1,318 @@
|
||||
import typing as t
|
||||
|
||||
from . import nodes
|
||||
from .visitor import NodeVisitor
|
||||
|
||||
if t.TYPE_CHECKING:
|
||||
import typing_extensions as te
|
||||
|
||||
VAR_LOAD_PARAMETER = "param"
|
||||
VAR_LOAD_RESOLVE = "resolve"
|
||||
VAR_LOAD_ALIAS = "alias"
|
||||
VAR_LOAD_UNDEFINED = "undefined"
|
||||
|
||||
|
||||
def find_symbols(
|
||||
nodes: t.Iterable[nodes.Node], parent_symbols: t.Optional["Symbols"] = None
|
||||
) -> "Symbols":
|
||||
sym = Symbols(parent=parent_symbols)
|
||||
visitor = FrameSymbolVisitor(sym)
|
||||
for node in nodes:
|
||||
visitor.visit(node)
|
||||
return sym
|
||||
|
||||
|
||||
def symbols_for_node(
|
||||
node: nodes.Node, parent_symbols: t.Optional["Symbols"] = None
|
||||
) -> "Symbols":
|
||||
sym = Symbols(parent=parent_symbols)
|
||||
sym.analyze_node(node)
|
||||
return sym
|
||||
|
||||
|
||||
class Symbols:
|
||||
def __init__(
|
||||
self, parent: t.Optional["Symbols"] = None, level: t.Optional[int] = None
|
||||
) -> None:
|
||||
if level is None:
|
||||
if parent is None:
|
||||
level = 0
|
||||
else:
|
||||
level = parent.level + 1
|
||||
|
||||
self.level: int = level
|
||||
self.parent = parent
|
||||
self.refs: t.Dict[str, str] = {}
|
||||
self.loads: t.Dict[str, t.Any] = {}
|
||||
self.stores: t.Set[str] = set()
|
||||
|
||||
def analyze_node(self, node: nodes.Node, **kwargs: t.Any) -> None:
|
||||
visitor = RootVisitor(self)
|
||||
visitor.visit(node, **kwargs)
|
||||
|
||||
def _define_ref(
|
||||
self, name: str, load: t.Optional[t.Tuple[str, t.Optional[str]]] = None
|
||||
) -> str:
|
||||
ident = f"l_{self.level}_{name}"
|
||||
self.refs[name] = ident
|
||||
if load is not None:
|
||||
self.loads[ident] = load
|
||||
return ident
|
||||
|
||||
def find_load(self, target: str) -> t.Optional[t.Any]:
|
||||
if target in self.loads:
|
||||
return self.loads[target]
|
||||
|
||||
if self.parent is not None:
|
||||
return self.parent.find_load(target)
|
||||
|
||||
return None
|
||||
|
||||
def find_ref(self, name: str) -> t.Optional[str]:
|
||||
if name in self.refs:
|
||||
return self.refs[name]
|
||||
|
||||
if self.parent is not None:
|
||||
return self.parent.find_ref(name)
|
||||
|
||||
return None
|
||||
|
||||
def ref(self, name: str) -> str:
|
||||
rv = self.find_ref(name)
|
||||
if rv is None:
|
||||
raise AssertionError(
|
||||
"Tried to resolve a name to a reference that was"
|
||||
f" unknown to the frame ({name!r})"
|
||||
)
|
||||
return rv
|
||||
|
||||
def copy(self) -> "te.Self":
|
||||
rv = object.__new__(self.__class__)
|
||||
rv.__dict__.update(self.__dict__)
|
||||
rv.refs = self.refs.copy()
|
||||
rv.loads = self.loads.copy()
|
||||
rv.stores = self.stores.copy()
|
||||
return rv
|
||||
|
||||
def store(self, name: str) -> None:
|
||||
self.stores.add(name)
|
||||
|
||||
# If we have not see the name referenced yet, we need to figure
|
||||
# out what to set it to.
|
||||
if name not in self.refs:
|
||||
# If there is a parent scope we check if the name has a
|
||||
# reference there. If it does it means we might have to alias
|
||||
# to a variable there.
|
||||
if self.parent is not None:
|
||||
outer_ref = self.parent.find_ref(name)
|
||||
if outer_ref is not None:
|
||||
self._define_ref(name, load=(VAR_LOAD_ALIAS, outer_ref))
|
||||
return
|
||||
|
||||
# Otherwise we can just set it to undefined.
|
||||
self._define_ref(name, load=(VAR_LOAD_UNDEFINED, None))
|
||||
|
||||
def declare_parameter(self, name: str) -> str:
|
||||
self.stores.add(name)
|
||||
return self._define_ref(name, load=(VAR_LOAD_PARAMETER, None))
|
||||
|
||||
def load(self, name: str) -> None:
|
||||
if self.find_ref(name) is None:
|
||||
self._define_ref(name, load=(VAR_LOAD_RESOLVE, name))
|
||||
|
||||
def branch_update(self, branch_symbols: t.Sequence["Symbols"]) -> None:
|
||||
stores: t.Set[str] = set()
|
||||
|
||||
for branch in branch_symbols:
|
||||
stores.update(branch.stores)
|
||||
|
||||
stores.difference_update(self.stores)
|
||||
|
||||
for sym in branch_symbols:
|
||||
self.refs.update(sym.refs)
|
||||
self.loads.update(sym.loads)
|
||||
self.stores.update(sym.stores)
|
||||
|
||||
for name in stores:
|
||||
target = self.find_ref(name)
|
||||
assert target is not None, "should not happen"
|
||||
|
||||
if self.parent is not None:
|
||||
outer_target = self.parent.find_ref(name)
|
||||
if outer_target is not None:
|
||||
self.loads[target] = (VAR_LOAD_ALIAS, outer_target)
|
||||
continue
|
||||
self.loads[target] = (VAR_LOAD_RESOLVE, name)
|
||||
|
||||
def dump_stores(self) -> t.Dict[str, str]:
|
||||
rv: t.Dict[str, str] = {}
|
||||
node: t.Optional[Symbols] = self
|
||||
|
||||
while node is not None:
|
||||
for name in sorted(node.stores):
|
||||
if name not in rv:
|
||||
rv[name] = self.find_ref(name) # type: ignore
|
||||
|
||||
node = node.parent
|
||||
|
||||
return rv
|
||||
|
||||
def dump_param_targets(self) -> t.Set[str]:
|
||||
rv = set()
|
||||
node: t.Optional[Symbols] = self
|
||||
|
||||
while node is not None:
|
||||
for target, (instr, _) in self.loads.items():
|
||||
if instr == VAR_LOAD_PARAMETER:
|
||||
rv.add(target)
|
||||
|
||||
node = node.parent
|
||||
|
||||
return rv
|
||||
|
||||
|
||||
class RootVisitor(NodeVisitor):
|
||||
def __init__(self, symbols: "Symbols") -> None:
|
||||
self.sym_visitor = FrameSymbolVisitor(symbols)
|
||||
|
||||
def _simple_visit(self, node: nodes.Node, **kwargs: t.Any) -> None:
|
||||
for child in node.iter_child_nodes():
|
||||
self.sym_visitor.visit(child)
|
||||
|
||||
visit_Template = _simple_visit
|
||||
visit_Block = _simple_visit
|
||||
visit_Macro = _simple_visit
|
||||
visit_FilterBlock = _simple_visit
|
||||
visit_Scope = _simple_visit
|
||||
visit_If = _simple_visit
|
||||
visit_ScopedEvalContextModifier = _simple_visit
|
||||
|
||||
def visit_AssignBlock(self, node: nodes.AssignBlock, **kwargs: t.Any) -> None:
|
||||
for child in node.body:
|
||||
self.sym_visitor.visit(child)
|
||||
|
||||
def visit_CallBlock(self, node: nodes.CallBlock, **kwargs: t.Any) -> None:
|
||||
for child in node.iter_child_nodes(exclude=("call",)):
|
||||
self.sym_visitor.visit(child)
|
||||
|
||||
def visit_OverlayScope(self, node: nodes.OverlayScope, **kwargs: t.Any) -> None:
|
||||
for child in node.body:
|
||||
self.sym_visitor.visit(child)
|
||||
|
||||
def visit_For(
|
||||
self, node: nodes.For, for_branch: str = "body", **kwargs: t.Any
|
||||
) -> None:
|
||||
if for_branch == "body":
|
||||
self.sym_visitor.visit(node.target, store_as_param=True)
|
||||
branch = node.body
|
||||
elif for_branch == "else":
|
||||
branch = node.else_
|
||||
elif for_branch == "test":
|
||||
self.sym_visitor.visit(node.target, store_as_param=True)
|
||||
if node.test is not None:
|
||||
self.sym_visitor.visit(node.test)
|
||||
return
|
||||
else:
|
||||
raise RuntimeError("Unknown for branch")
|
||||
|
||||
if branch:
|
||||
for item in branch:
|
||||
self.sym_visitor.visit(item)
|
||||
|
||||
def visit_With(self, node: nodes.With, **kwargs: t.Any) -> None:
|
||||
for target in node.targets:
|
||||
self.sym_visitor.visit(target)
|
||||
for child in node.body:
|
||||
self.sym_visitor.visit(child)
|
||||
|
||||
def generic_visit(self, node: nodes.Node, *args: t.Any, **kwargs: t.Any) -> None:
|
||||
raise NotImplementedError(f"Cannot find symbols for {type(node).__name__!r}")
|
||||
|
||||
|
||||
class FrameSymbolVisitor(NodeVisitor):
|
||||
"""A visitor for `Frame.inspect`."""
|
||||
|
||||
def __init__(self, symbols: "Symbols") -> None:
|
||||
self.symbols = symbols
|
||||
|
||||
def visit_Name(
|
||||
self, node: nodes.Name, store_as_param: bool = False, **kwargs: t.Any
|
||||
) -> None:
|
||||
"""All assignments to names go through this function."""
|
||||
if store_as_param or node.ctx == "param":
|
||||
self.symbols.declare_parameter(node.name)
|
||||
elif node.ctx == "store":
|
||||
self.symbols.store(node.name)
|
||||
elif node.ctx == "load":
|
||||
self.symbols.load(node.name)
|
||||
|
||||
def visit_NSRef(self, node: nodes.NSRef, **kwargs: t.Any) -> None:
|
||||
self.symbols.load(node.name)
|
||||
|
||||
def visit_If(self, node: nodes.If, **kwargs: t.Any) -> None:
|
||||
self.visit(node.test, **kwargs)
|
||||
original_symbols = self.symbols
|
||||
|
||||
def inner_visit(nodes: t.Iterable[nodes.Node]) -> "Symbols":
|
||||
self.symbols = rv = original_symbols.copy()
|
||||
|
||||
for subnode in nodes:
|
||||
self.visit(subnode, **kwargs)
|
||||
|
||||
self.symbols = original_symbols
|
||||
return rv
|
||||
|
||||
body_symbols = inner_visit(node.body)
|
||||
elif_symbols = inner_visit(node.elif_)
|
||||
else_symbols = inner_visit(node.else_ or ())
|
||||
self.symbols.branch_update([body_symbols, elif_symbols, else_symbols])
|
||||
|
||||
def visit_Macro(self, node: nodes.Macro, **kwargs: t.Any) -> None:
|
||||
self.symbols.store(node.name)
|
||||
|
||||
def visit_Import(self, node: nodes.Import, **kwargs: t.Any) -> None:
|
||||
self.generic_visit(node, **kwargs)
|
||||
self.symbols.store(node.target)
|
||||
|
||||
def visit_FromImport(self, node: nodes.FromImport, **kwargs: t.Any) -> None:
|
||||
self.generic_visit(node, **kwargs)
|
||||
|
||||
for name in node.names:
|
||||
if isinstance(name, tuple):
|
||||
self.symbols.store(name[1])
|
||||
else:
|
||||
self.symbols.store(name)
|
||||
|
||||
def visit_Assign(self, node: nodes.Assign, **kwargs: t.Any) -> None:
|
||||
"""Visit assignments in the correct order."""
|
||||
self.visit(node.node, **kwargs)
|
||||
self.visit(node.target, **kwargs)
|
||||
|
||||
def visit_For(self, node: nodes.For, **kwargs: t.Any) -> None:
|
||||
"""Visiting stops at for blocks. However the block sequence
|
||||
is visited as part of the outer scope.
|
||||
"""
|
||||
self.visit(node.iter, **kwargs)
|
||||
|
||||
def visit_CallBlock(self, node: nodes.CallBlock, **kwargs: t.Any) -> None:
|
||||
self.visit(node.call, **kwargs)
|
||||
|
||||
def visit_FilterBlock(self, node: nodes.FilterBlock, **kwargs: t.Any) -> None:
|
||||
self.visit(node.filter, **kwargs)
|
||||
|
||||
def visit_With(self, node: nodes.With, **kwargs: t.Any) -> None:
|
||||
for target in node.values:
|
||||
self.visit(target)
|
||||
|
||||
def visit_AssignBlock(self, node: nodes.AssignBlock, **kwargs: t.Any) -> None:
|
||||
"""Stop visiting at block assigns."""
|
||||
self.visit(node.target, **kwargs)
|
||||
|
||||
def visit_Scope(self, node: nodes.Scope, **kwargs: t.Any) -> None:
|
||||
"""Stop visiting at scopes."""
|
||||
|
||||
def visit_Block(self, node: nodes.Block, **kwargs: t.Any) -> None:
|
||||
"""Stop visiting at blocks."""
|
||||
|
||||
def visit_OverlayScope(self, node: nodes.OverlayScope, **kwargs: t.Any) -> None:
|
||||
"""Do not visit into overlay scopes."""
|
||||
868
venv/Lib/site-packages/jinja2/lexer.py
Normal file
868
venv/Lib/site-packages/jinja2/lexer.py
Normal file
@@ -0,0 +1,868 @@
|
||||
"""Implements a Jinja / Python combination lexer. The ``Lexer`` class
|
||||
is used to do some preprocessing. It filters out invalid operators like
|
||||
the bitshift operators we don't allow in templates. It separates
|
||||
template code and python code in expressions.
|
||||
"""
|
||||
|
||||
import re
|
||||
import typing as t
|
||||
from ast import literal_eval
|
||||
from collections import deque
|
||||
from sys import intern
|
||||
|
||||
from ._identifier import pattern as name_re
|
||||
from .exceptions import TemplateSyntaxError
|
||||
from .utils import LRUCache
|
||||
|
||||
if t.TYPE_CHECKING:
|
||||
import typing_extensions as te
|
||||
|
||||
from .environment import Environment
|
||||
|
||||
# cache for the lexers. Exists in order to be able to have multiple
|
||||
# environments with the same lexer
|
||||
_lexer_cache: t.MutableMapping[t.Tuple, "Lexer"] = LRUCache(50) # type: ignore
|
||||
|
||||
# static regular expressions
|
||||
whitespace_re = re.compile(r"\s+")
|
||||
newline_re = re.compile(r"(\r\n|\r|\n)")
|
||||
string_re = re.compile(
|
||||
r"('([^'\\]*(?:\\.[^'\\]*)*)'" r'|"([^"\\]*(?:\\.[^"\\]*)*)")', re.S
|
||||
)
|
||||
integer_re = re.compile(
|
||||
r"""
|
||||
(
|
||||
0b(_?[0-1])+ # binary
|
||||
|
|
||||
0o(_?[0-7])+ # octal
|
||||
|
|
||||
0x(_?[\da-f])+ # hex
|
||||
|
|
||||
[1-9](_?\d)* # decimal
|
||||
|
|
||||
0(_?0)* # decimal zero
|
||||
)
|
||||
""",
|
||||
re.IGNORECASE | re.VERBOSE,
|
||||
)
|
||||
float_re = re.compile(
|
||||
r"""
|
||||
(?<!\.) # doesn't start with a .
|
||||
(\d+_)*\d+ # digits, possibly _ separated
|
||||
(
|
||||
(\.(\d+_)*\d+)? # optional fractional part
|
||||
e[+\-]?(\d+_)*\d+ # exponent part
|
||||
|
|
||||
\.(\d+_)*\d+ # required fractional part
|
||||
)
|
||||
""",
|
||||
re.IGNORECASE | re.VERBOSE,
|
||||
)
|
||||
|
||||
# internal the tokens and keep references to them
|
||||
TOKEN_ADD = intern("add")
|
||||
TOKEN_ASSIGN = intern("assign")
|
||||
TOKEN_COLON = intern("colon")
|
||||
TOKEN_COMMA = intern("comma")
|
||||
TOKEN_DIV = intern("div")
|
||||
TOKEN_DOT = intern("dot")
|
||||
TOKEN_EQ = intern("eq")
|
||||
TOKEN_FLOORDIV = intern("floordiv")
|
||||
TOKEN_GT = intern("gt")
|
||||
TOKEN_GTEQ = intern("gteq")
|
||||
TOKEN_LBRACE = intern("lbrace")
|
||||
TOKEN_LBRACKET = intern("lbracket")
|
||||
TOKEN_LPAREN = intern("lparen")
|
||||
TOKEN_LT = intern("lt")
|
||||
TOKEN_LTEQ = intern("lteq")
|
||||
TOKEN_MOD = intern("mod")
|
||||
TOKEN_MUL = intern("mul")
|
||||
TOKEN_NE = intern("ne")
|
||||
TOKEN_PIPE = intern("pipe")
|
||||
TOKEN_POW = intern("pow")
|
||||
TOKEN_RBRACE = intern("rbrace")
|
||||
TOKEN_RBRACKET = intern("rbracket")
|
||||
TOKEN_RPAREN = intern("rparen")
|
||||
TOKEN_SEMICOLON = intern("semicolon")
|
||||
TOKEN_SUB = intern("sub")
|
||||
TOKEN_TILDE = intern("tilde")
|
||||
TOKEN_WHITESPACE = intern("whitespace")
|
||||
TOKEN_FLOAT = intern("float")
|
||||
TOKEN_INTEGER = intern("integer")
|
||||
TOKEN_NAME = intern("name")
|
||||
TOKEN_STRING = intern("string")
|
||||
TOKEN_OPERATOR = intern("operator")
|
||||
TOKEN_BLOCK_BEGIN = intern("block_begin")
|
||||
TOKEN_BLOCK_END = intern("block_end")
|
||||
TOKEN_VARIABLE_BEGIN = intern("variable_begin")
|
||||
TOKEN_VARIABLE_END = intern("variable_end")
|
||||
TOKEN_RAW_BEGIN = intern("raw_begin")
|
||||
TOKEN_RAW_END = intern("raw_end")
|
||||
TOKEN_COMMENT_BEGIN = intern("comment_begin")
|
||||
TOKEN_COMMENT_END = intern("comment_end")
|
||||
TOKEN_COMMENT = intern("comment")
|
||||
TOKEN_LINESTATEMENT_BEGIN = intern("linestatement_begin")
|
||||
TOKEN_LINESTATEMENT_END = intern("linestatement_end")
|
||||
TOKEN_LINECOMMENT_BEGIN = intern("linecomment_begin")
|
||||
TOKEN_LINECOMMENT_END = intern("linecomment_end")
|
||||
TOKEN_LINECOMMENT = intern("linecomment")
|
||||
TOKEN_DATA = intern("data")
|
||||
TOKEN_INITIAL = intern("initial")
|
||||
TOKEN_EOF = intern("eof")
|
||||
|
||||
# bind operators to token types
|
||||
operators = {
|
||||
"+": TOKEN_ADD,
|
||||
"-": TOKEN_SUB,
|
||||
"/": TOKEN_DIV,
|
||||
"//": TOKEN_FLOORDIV,
|
||||
"*": TOKEN_MUL,
|
||||
"%": TOKEN_MOD,
|
||||
"**": TOKEN_POW,
|
||||
"~": TOKEN_TILDE,
|
||||
"[": TOKEN_LBRACKET,
|
||||
"]": TOKEN_RBRACKET,
|
||||
"(": TOKEN_LPAREN,
|
||||
")": TOKEN_RPAREN,
|
||||
"{": TOKEN_LBRACE,
|
||||
"}": TOKEN_RBRACE,
|
||||
"==": TOKEN_EQ,
|
||||
"!=": TOKEN_NE,
|
||||
">": TOKEN_GT,
|
||||
">=": TOKEN_GTEQ,
|
||||
"<": TOKEN_LT,
|
||||
"<=": TOKEN_LTEQ,
|
||||
"=": TOKEN_ASSIGN,
|
||||
".": TOKEN_DOT,
|
||||
":": TOKEN_COLON,
|
||||
"|": TOKEN_PIPE,
|
||||
",": TOKEN_COMMA,
|
||||
";": TOKEN_SEMICOLON,
|
||||
}
|
||||
|
||||
reverse_operators = {v: k for k, v in operators.items()}
|
||||
assert len(operators) == len(reverse_operators), "operators dropped"
|
||||
operator_re = re.compile(
|
||||
f"({'|'.join(re.escape(x) for x in sorted(operators, key=lambda x: -len(x)))})"
|
||||
)
|
||||
|
||||
ignored_tokens = frozenset(
|
||||
[
|
||||
TOKEN_COMMENT_BEGIN,
|
||||
TOKEN_COMMENT,
|
||||
TOKEN_COMMENT_END,
|
||||
TOKEN_WHITESPACE,
|
||||
TOKEN_LINECOMMENT_BEGIN,
|
||||
TOKEN_LINECOMMENT_END,
|
||||
TOKEN_LINECOMMENT,
|
||||
]
|
||||
)
|
||||
ignore_if_empty = frozenset(
|
||||
[TOKEN_WHITESPACE, TOKEN_DATA, TOKEN_COMMENT, TOKEN_LINECOMMENT]
|
||||
)
|
||||
|
||||
|
||||
def _describe_token_type(token_type: str) -> str:
|
||||
if token_type in reverse_operators:
|
||||
return reverse_operators[token_type]
|
||||
|
||||
return {
|
||||
TOKEN_COMMENT_BEGIN: "begin of comment",
|
||||
TOKEN_COMMENT_END: "end of comment",
|
||||
TOKEN_COMMENT: "comment",
|
||||
TOKEN_LINECOMMENT: "comment",
|
||||
TOKEN_BLOCK_BEGIN: "begin of statement block",
|
||||
TOKEN_BLOCK_END: "end of statement block",
|
||||
TOKEN_VARIABLE_BEGIN: "begin of print statement",
|
||||
TOKEN_VARIABLE_END: "end of print statement",
|
||||
TOKEN_LINESTATEMENT_BEGIN: "begin of line statement",
|
||||
TOKEN_LINESTATEMENT_END: "end of line statement",
|
||||
TOKEN_DATA: "template data / text",
|
||||
TOKEN_EOF: "end of template",
|
||||
}.get(token_type, token_type)
|
||||
|
||||
|
||||
def describe_token(token: "Token") -> str:
|
||||
"""Returns a description of the token."""
|
||||
if token.type == TOKEN_NAME:
|
||||
return token.value
|
||||
|
||||
return _describe_token_type(token.type)
|
||||
|
||||
|
||||
def describe_token_expr(expr: str) -> str:
|
||||
"""Like `describe_token` but for token expressions."""
|
||||
if ":" in expr:
|
||||
type, value = expr.split(":", 1)
|
||||
|
||||
if type == TOKEN_NAME:
|
||||
return value
|
||||
else:
|
||||
type = expr
|
||||
|
||||
return _describe_token_type(type)
|
||||
|
||||
|
||||
def count_newlines(value: str) -> int:
|
||||
"""Count the number of newline characters in the string. This is
|
||||
useful for extensions that filter a stream.
|
||||
"""
|
||||
return len(newline_re.findall(value))
|
||||
|
||||
|
||||
def compile_rules(environment: "Environment") -> t.List[t.Tuple[str, str]]:
|
||||
"""Compiles all the rules from the environment into a list of rules."""
|
||||
e = re.escape
|
||||
rules = [
|
||||
(
|
||||
len(environment.comment_start_string),
|
||||
TOKEN_COMMENT_BEGIN,
|
||||
e(environment.comment_start_string),
|
||||
),
|
||||
(
|
||||
len(environment.block_start_string),
|
||||
TOKEN_BLOCK_BEGIN,
|
||||
e(environment.block_start_string),
|
||||
),
|
||||
(
|
||||
len(environment.variable_start_string),
|
||||
TOKEN_VARIABLE_BEGIN,
|
||||
e(environment.variable_start_string),
|
||||
),
|
||||
]
|
||||
|
||||
if environment.line_statement_prefix is not None:
|
||||
rules.append(
|
||||
(
|
||||
len(environment.line_statement_prefix),
|
||||
TOKEN_LINESTATEMENT_BEGIN,
|
||||
r"^[ \t\v]*" + e(environment.line_statement_prefix),
|
||||
)
|
||||
)
|
||||
if environment.line_comment_prefix is not None:
|
||||
rules.append(
|
||||
(
|
||||
len(environment.line_comment_prefix),
|
||||
TOKEN_LINECOMMENT_BEGIN,
|
||||
r"(?:^|(?<=\S))[^\S\r\n]*" + e(environment.line_comment_prefix),
|
||||
)
|
||||
)
|
||||
|
||||
return [x[1:] for x in sorted(rules, reverse=True)]
|
||||
|
||||
|
||||
class Failure:
|
||||
"""Class that raises a `TemplateSyntaxError` if called.
|
||||
Used by the `Lexer` to specify known errors.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self, message: str, cls: t.Type[TemplateSyntaxError] = TemplateSyntaxError
|
||||
) -> None:
|
||||
self.message = message
|
||||
self.error_class = cls
|
||||
|
||||
def __call__(self, lineno: int, filename: t.Optional[str]) -> "te.NoReturn":
|
||||
raise self.error_class(self.message, lineno, filename)
|
||||
|
||||
|
||||
class Token(t.NamedTuple):
|
||||
lineno: int
|
||||
type: str
|
||||
value: str
|
||||
|
||||
def __str__(self) -> str:
|
||||
return describe_token(self)
|
||||
|
||||
def test(self, expr: str) -> bool:
|
||||
"""Test a token against a token expression. This can either be a
|
||||
token type or ``'token_type:token_value'``. This can only test
|
||||
against string values and types.
|
||||
"""
|
||||
# here we do a regular string equality check as test_any is usually
|
||||
# passed an iterable of not interned strings.
|
||||
if self.type == expr:
|
||||
return True
|
||||
|
||||
if ":" in expr:
|
||||
return expr.split(":", 1) == [self.type, self.value]
|
||||
|
||||
return False
|
||||
|
||||
def test_any(self, *iterable: str) -> bool:
|
||||
"""Test against multiple token expressions."""
|
||||
return any(self.test(expr) for expr in iterable)
|
||||
|
||||
|
||||
class TokenStreamIterator:
|
||||
"""The iterator for tokenstreams. Iterate over the stream
|
||||
until the eof token is reached.
|
||||
"""
|
||||
|
||||
def __init__(self, stream: "TokenStream") -> None:
|
||||
self.stream = stream
|
||||
|
||||
def __iter__(self) -> "TokenStreamIterator":
|
||||
return self
|
||||
|
||||
def __next__(self) -> Token:
|
||||
token = self.stream.current
|
||||
|
||||
if token.type is TOKEN_EOF:
|
||||
self.stream.close()
|
||||
raise StopIteration
|
||||
|
||||
next(self.stream)
|
||||
return token
|
||||
|
||||
|
||||
class TokenStream:
|
||||
"""A token stream is an iterable that yields :class:`Token`\\s. The
|
||||
parser however does not iterate over it but calls :meth:`next` to go
|
||||
one token ahead. The current active token is stored as :attr:`current`.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
generator: t.Iterable[Token],
|
||||
name: t.Optional[str],
|
||||
filename: t.Optional[str],
|
||||
):
|
||||
self._iter = iter(generator)
|
||||
self._pushed: te.Deque[Token] = deque()
|
||||
self.name = name
|
||||
self.filename = filename
|
||||
self.closed = False
|
||||
self.current = Token(1, TOKEN_INITIAL, "")
|
||||
next(self)
|
||||
|
||||
def __iter__(self) -> TokenStreamIterator:
|
||||
return TokenStreamIterator(self)
|
||||
|
||||
def __bool__(self) -> bool:
|
||||
return bool(self._pushed) or self.current.type is not TOKEN_EOF
|
||||
|
||||
@property
|
||||
def eos(self) -> bool:
|
||||
"""Are we at the end of the stream?"""
|
||||
return not self
|
||||
|
||||
def push(self, token: Token) -> None:
|
||||
"""Push a token back to the stream."""
|
||||
self._pushed.append(token)
|
||||
|
||||
def look(self) -> Token:
|
||||
"""Look at the next token."""
|
||||
old_token = next(self)
|
||||
result = self.current
|
||||
self.push(result)
|
||||
self.current = old_token
|
||||
return result
|
||||
|
||||
def skip(self, n: int = 1) -> None:
|
||||
"""Got n tokens ahead."""
|
||||
for _ in range(n):
|
||||
next(self)
|
||||
|
||||
def next_if(self, expr: str) -> t.Optional[Token]:
|
||||
"""Perform the token test and return the token if it matched.
|
||||
Otherwise the return value is `None`.
|
||||
"""
|
||||
if self.current.test(expr):
|
||||
return next(self)
|
||||
|
||||
return None
|
||||
|
||||
def skip_if(self, expr: str) -> bool:
|
||||
"""Like :meth:`next_if` but only returns `True` or `False`."""
|
||||
return self.next_if(expr) is not None
|
||||
|
||||
def __next__(self) -> Token:
|
||||
"""Go one token ahead and return the old one.
|
||||
|
||||
Use the built-in :func:`next` instead of calling this directly.
|
||||
"""
|
||||
rv = self.current
|
||||
|
||||
if self._pushed:
|
||||
self.current = self._pushed.popleft()
|
||||
elif self.current.type is not TOKEN_EOF:
|
||||
try:
|
||||
self.current = next(self._iter)
|
||||
except StopIteration:
|
||||
self.close()
|
||||
|
||||
return rv
|
||||
|
||||
def close(self) -> None:
|
||||
"""Close the stream."""
|
||||
self.current = Token(self.current.lineno, TOKEN_EOF, "")
|
||||
self._iter = iter(())
|
||||
self.closed = True
|
||||
|
||||
def expect(self, expr: str) -> Token:
|
||||
"""Expect a given token type and return it. This accepts the same
|
||||
argument as :meth:`jinja2.lexer.Token.test`.
|
||||
"""
|
||||
if not self.current.test(expr):
|
||||
expr = describe_token_expr(expr)
|
||||
|
||||
if self.current.type is TOKEN_EOF:
|
||||
raise TemplateSyntaxError(
|
||||
f"unexpected end of template, expected {expr!r}.",
|
||||
self.current.lineno,
|
||||
self.name,
|
||||
self.filename,
|
||||
)
|
||||
|
||||
raise TemplateSyntaxError(
|
||||
f"expected token {expr!r}, got {describe_token(self.current)!r}",
|
||||
self.current.lineno,
|
||||
self.name,
|
||||
self.filename,
|
||||
)
|
||||
|
||||
return next(self)
|
||||
|
||||
|
||||
def get_lexer(environment: "Environment") -> "Lexer":
|
||||
"""Return a lexer which is probably cached."""
|
||||
key = (
|
||||
environment.block_start_string,
|
||||
environment.block_end_string,
|
||||
environment.variable_start_string,
|
||||
environment.variable_end_string,
|
||||
environment.comment_start_string,
|
||||
environment.comment_end_string,
|
||||
environment.line_statement_prefix,
|
||||
environment.line_comment_prefix,
|
||||
environment.trim_blocks,
|
||||
environment.lstrip_blocks,
|
||||
environment.newline_sequence,
|
||||
environment.keep_trailing_newline,
|
||||
)
|
||||
lexer = _lexer_cache.get(key)
|
||||
|
||||
if lexer is None:
|
||||
_lexer_cache[key] = lexer = Lexer(environment)
|
||||
|
||||
return lexer
|
||||
|
||||
|
||||
class OptionalLStrip(tuple): # type: ignore[type-arg]
|
||||
"""A special tuple for marking a point in the state that can have
|
||||
lstrip applied.
|
||||
"""
|
||||
|
||||
__slots__ = ()
|
||||
|
||||
# Even though it looks like a no-op, creating instances fails
|
||||
# without this.
|
||||
def __new__(cls, *members, **kwargs): # type: ignore
|
||||
return super().__new__(cls, members)
|
||||
|
||||
|
||||
class _Rule(t.NamedTuple):
|
||||
pattern: t.Pattern[str]
|
||||
tokens: t.Union[str, t.Tuple[str, ...], t.Tuple[Failure]]
|
||||
command: t.Optional[str]
|
||||
|
||||
|
||||
class Lexer:
|
||||
"""Class that implements a lexer for a given environment. Automatically
|
||||
created by the environment class, usually you don't have to do that.
|
||||
|
||||
Note that the lexer is not automatically bound to an environment.
|
||||
Multiple environments can share the same lexer.
|
||||
"""
|
||||
|
||||
def __init__(self, environment: "Environment") -> None:
|
||||
# shortcuts
|
||||
e = re.escape
|
||||
|
||||
def c(x: str) -> t.Pattern[str]:
|
||||
return re.compile(x, re.M | re.S)
|
||||
|
||||
# lexing rules for tags
|
||||
tag_rules: t.List[_Rule] = [
|
||||
_Rule(whitespace_re, TOKEN_WHITESPACE, None),
|
||||
_Rule(float_re, TOKEN_FLOAT, None),
|
||||
_Rule(integer_re, TOKEN_INTEGER, None),
|
||||
_Rule(name_re, TOKEN_NAME, None),
|
||||
_Rule(string_re, TOKEN_STRING, None),
|
||||
_Rule(operator_re, TOKEN_OPERATOR, None),
|
||||
]
|
||||
|
||||
# assemble the root lexing rule. because "|" is ungreedy
|
||||
# we have to sort by length so that the lexer continues working
|
||||
# as expected when we have parsing rules like <% for block and
|
||||
# <%= for variables. (if someone wants asp like syntax)
|
||||
# variables are just part of the rules if variable processing
|
||||
# is required.
|
||||
root_tag_rules = compile_rules(environment)
|
||||
|
||||
block_start_re = e(environment.block_start_string)
|
||||
block_end_re = e(environment.block_end_string)
|
||||
comment_end_re = e(environment.comment_end_string)
|
||||
variable_end_re = e(environment.variable_end_string)
|
||||
|
||||
# block suffix if trimming is enabled
|
||||
block_suffix_re = "\\n?" if environment.trim_blocks else ""
|
||||
|
||||
self.lstrip_blocks = environment.lstrip_blocks
|
||||
|
||||
self.newline_sequence = environment.newline_sequence
|
||||
self.keep_trailing_newline = environment.keep_trailing_newline
|
||||
|
||||
root_raw_re = (
|
||||
rf"(?P<raw_begin>{block_start_re}(\-|\+|)\s*raw\s*"
|
||||
rf"(?:\-{block_end_re}\s*|{block_end_re}))"
|
||||
)
|
||||
root_parts_re = "|".join(
|
||||
[root_raw_re] + [rf"(?P<{n}>{r}(\-|\+|))" for n, r in root_tag_rules]
|
||||
)
|
||||
|
||||
# global lexing rules
|
||||
self.rules: t.Dict[str, t.List[_Rule]] = {
|
||||
"root": [
|
||||
# directives
|
||||
_Rule(
|
||||
c(rf"(.*?)(?:{root_parts_re})"),
|
||||
OptionalLStrip(TOKEN_DATA, "#bygroup"), # type: ignore
|
||||
"#bygroup",
|
||||
),
|
||||
# data
|
||||
_Rule(c(".+"), TOKEN_DATA, None),
|
||||
],
|
||||
# comments
|
||||
TOKEN_COMMENT_BEGIN: [
|
||||
_Rule(
|
||||
c(
|
||||
rf"(.*?)((?:\+{comment_end_re}|\-{comment_end_re}\s*"
|
||||
rf"|{comment_end_re}{block_suffix_re}))"
|
||||
),
|
||||
(TOKEN_COMMENT, TOKEN_COMMENT_END),
|
||||
"#pop",
|
||||
),
|
||||
_Rule(c(r"(.)"), (Failure("Missing end of comment tag"),), None),
|
||||
],
|
||||
# blocks
|
||||
TOKEN_BLOCK_BEGIN: [
|
||||
_Rule(
|
||||
c(
|
||||
rf"(?:\+{block_end_re}|\-{block_end_re}\s*"
|
||||
rf"|{block_end_re}{block_suffix_re})"
|
||||
),
|
||||
TOKEN_BLOCK_END,
|
||||
"#pop",
|
||||
),
|
||||
]
|
||||
+ tag_rules,
|
||||
# variables
|
||||
TOKEN_VARIABLE_BEGIN: [
|
||||
_Rule(
|
||||
c(rf"\-{variable_end_re}\s*|{variable_end_re}"),
|
||||
TOKEN_VARIABLE_END,
|
||||
"#pop",
|
||||
)
|
||||
]
|
||||
+ tag_rules,
|
||||
# raw block
|
||||
TOKEN_RAW_BEGIN: [
|
||||
_Rule(
|
||||
c(
|
||||
rf"(.*?)((?:{block_start_re}(\-|\+|))\s*endraw\s*"
|
||||
rf"(?:\+{block_end_re}|\-{block_end_re}\s*"
|
||||
rf"|{block_end_re}{block_suffix_re}))"
|
||||
),
|
||||
OptionalLStrip(TOKEN_DATA, TOKEN_RAW_END), # type: ignore
|
||||
"#pop",
|
||||
),
|
||||
_Rule(c(r"(.)"), (Failure("Missing end of raw directive"),), None),
|
||||
],
|
||||
# line statements
|
||||
TOKEN_LINESTATEMENT_BEGIN: [
|
||||
_Rule(c(r"\s*(\n|$)"), TOKEN_LINESTATEMENT_END, "#pop")
|
||||
]
|
||||
+ tag_rules,
|
||||
# line comments
|
||||
TOKEN_LINECOMMENT_BEGIN: [
|
||||
_Rule(
|
||||
c(r"(.*?)()(?=\n|$)"),
|
||||
(TOKEN_LINECOMMENT, TOKEN_LINECOMMENT_END),
|
||||
"#pop",
|
||||
)
|
||||
],
|
||||
}
|
||||
|
||||
def _normalize_newlines(self, value: str) -> str:
|
||||
"""Replace all newlines with the configured sequence in strings
|
||||
and template data.
|
||||
"""
|
||||
return newline_re.sub(self.newline_sequence, value)
|
||||
|
||||
def tokenize(
|
||||
self,
|
||||
source: str,
|
||||
name: t.Optional[str] = None,
|
||||
filename: t.Optional[str] = None,
|
||||
state: t.Optional[str] = None,
|
||||
) -> TokenStream:
|
||||
"""Calls tokeniter + tokenize and wraps it in a token stream."""
|
||||
stream = self.tokeniter(source, name, filename, state)
|
||||
return TokenStream(self.wrap(stream, name, filename), name, filename)
|
||||
|
||||
def wrap(
|
||||
self,
|
||||
stream: t.Iterable[t.Tuple[int, str, str]],
|
||||
name: t.Optional[str] = None,
|
||||
filename: t.Optional[str] = None,
|
||||
) -> t.Iterator[Token]:
|
||||
"""This is called with the stream as returned by `tokenize` and wraps
|
||||
every token in a :class:`Token` and converts the value.
|
||||
"""
|
||||
for lineno, token, value_str in stream:
|
||||
if token in ignored_tokens:
|
||||
continue
|
||||
|
||||
value: t.Any = value_str
|
||||
|
||||
if token == TOKEN_LINESTATEMENT_BEGIN:
|
||||
token = TOKEN_BLOCK_BEGIN
|
||||
elif token == TOKEN_LINESTATEMENT_END:
|
||||
token = TOKEN_BLOCK_END
|
||||
# we are not interested in those tokens in the parser
|
||||
elif token in (TOKEN_RAW_BEGIN, TOKEN_RAW_END):
|
||||
continue
|
||||
elif token == TOKEN_DATA:
|
||||
value = self._normalize_newlines(value_str)
|
||||
elif token == "keyword":
|
||||
token = value_str
|
||||
elif token == TOKEN_NAME:
|
||||
value = value_str
|
||||
|
||||
if not value.isidentifier():
|
||||
raise TemplateSyntaxError(
|
||||
"Invalid character in identifier", lineno, name, filename
|
||||
)
|
||||
elif token == TOKEN_STRING:
|
||||
# try to unescape string
|
||||
try:
|
||||
value = (
|
||||
self._normalize_newlines(value_str[1:-1])
|
||||
.encode("ascii", "backslashreplace")
|
||||
.decode("unicode-escape")
|
||||
)
|
||||
except Exception as e:
|
||||
msg = str(e).split(":")[-1].strip()
|
||||
raise TemplateSyntaxError(msg, lineno, name, filename) from e
|
||||
elif token == TOKEN_INTEGER:
|
||||
value = int(value_str.replace("_", ""), 0)
|
||||
elif token == TOKEN_FLOAT:
|
||||
# remove all "_" first to support more Python versions
|
||||
value = literal_eval(value_str.replace("_", ""))
|
||||
elif token == TOKEN_OPERATOR:
|
||||
token = operators[value_str]
|
||||
|
||||
yield Token(lineno, token, value)
|
||||
|
||||
def tokeniter(
|
||||
self,
|
||||
source: str,
|
||||
name: t.Optional[str],
|
||||
filename: t.Optional[str] = None,
|
||||
state: t.Optional[str] = None,
|
||||
) -> t.Iterator[t.Tuple[int, str, str]]:
|
||||
"""This method tokenizes the text and returns the tokens in a
|
||||
generator. Use this method if you just want to tokenize a template.
|
||||
|
||||
.. versionchanged:: 3.0
|
||||
Only ``\\n``, ``\\r\\n`` and ``\\r`` are treated as line
|
||||
breaks.
|
||||
"""
|
||||
lines = newline_re.split(source)[::2]
|
||||
|
||||
if not self.keep_trailing_newline and lines[-1] == "":
|
||||
del lines[-1]
|
||||
|
||||
source = "\n".join(lines)
|
||||
pos = 0
|
||||
lineno = 1
|
||||
stack = ["root"]
|
||||
|
||||
if state is not None and state != "root":
|
||||
assert state in ("variable", "block"), "invalid state"
|
||||
stack.append(state + "_begin")
|
||||
|
||||
statetokens = self.rules[stack[-1]]
|
||||
source_length = len(source)
|
||||
balancing_stack: t.List[str] = []
|
||||
newlines_stripped = 0
|
||||
line_starting = True
|
||||
|
||||
while True:
|
||||
# tokenizer loop
|
||||
for regex, tokens, new_state in statetokens:
|
||||
m = regex.match(source, pos)
|
||||
|
||||
# if no match we try again with the next rule
|
||||
if m is None:
|
||||
continue
|
||||
|
||||
# we only match blocks and variables if braces / parentheses
|
||||
# are balanced. continue parsing with the lower rule which
|
||||
# is the operator rule. do this only if the end tags look
|
||||
# like operators
|
||||
if balancing_stack and tokens in (
|
||||
TOKEN_VARIABLE_END,
|
||||
TOKEN_BLOCK_END,
|
||||
TOKEN_LINESTATEMENT_END,
|
||||
):
|
||||
continue
|
||||
|
||||
# tuples support more options
|
||||
if isinstance(tokens, tuple):
|
||||
groups: t.Sequence[str] = m.groups()
|
||||
|
||||
if isinstance(tokens, OptionalLStrip):
|
||||
# Rule supports lstrip. Match will look like
|
||||
# text, block type, whitespace control, type, control, ...
|
||||
text = groups[0]
|
||||
# Skipping the text and first type, every other group is the
|
||||
# whitespace control for each type. One of the groups will be
|
||||
# -, +, or empty string instead of None.
|
||||
strip_sign = next(g for g in groups[2::2] if g is not None)
|
||||
|
||||
if strip_sign == "-":
|
||||
# Strip all whitespace between the text and the tag.
|
||||
stripped = text.rstrip()
|
||||
newlines_stripped = text[len(stripped) :].count("\n")
|
||||
groups = [stripped, *groups[1:]]
|
||||
elif (
|
||||
# Not marked for preserving whitespace.
|
||||
strip_sign != "+"
|
||||
# lstrip is enabled.
|
||||
and self.lstrip_blocks
|
||||
# Not a variable expression.
|
||||
and not m.groupdict().get(TOKEN_VARIABLE_BEGIN)
|
||||
):
|
||||
# The start of text between the last newline and the tag.
|
||||
l_pos = text.rfind("\n") + 1
|
||||
|
||||
if l_pos > 0 or line_starting:
|
||||
# If there's only whitespace between the newline and the
|
||||
# tag, strip it.
|
||||
if whitespace_re.fullmatch(text, l_pos):
|
||||
groups = [text[:l_pos], *groups[1:]]
|
||||
|
||||
for idx, token in enumerate(tokens):
|
||||
# failure group
|
||||
if isinstance(token, Failure):
|
||||
raise token(lineno, filename)
|
||||
# bygroup is a bit more complex, in that case we
|
||||
# yield for the current token the first named
|
||||
# group that matched
|
||||
elif token == "#bygroup":
|
||||
for key, value in m.groupdict().items():
|
||||
if value is not None:
|
||||
yield lineno, key, value
|
||||
lineno += value.count("\n")
|
||||
break
|
||||
else:
|
||||
raise RuntimeError(
|
||||
f"{regex!r} wanted to resolve the token dynamically"
|
||||
" but no group matched"
|
||||
)
|
||||
# normal group
|
||||
else:
|
||||
data = groups[idx]
|
||||
|
||||
if data or token not in ignore_if_empty:
|
||||
yield lineno, token, data # type: ignore[misc]
|
||||
|
||||
lineno += data.count("\n") + newlines_stripped
|
||||
newlines_stripped = 0
|
||||
|
||||
# strings as token just are yielded as it.
|
||||
else:
|
||||
data = m.group()
|
||||
|
||||
# update brace/parentheses balance
|
||||
if tokens == TOKEN_OPERATOR:
|
||||
if data == "{":
|
||||
balancing_stack.append("}")
|
||||
elif data == "(":
|
||||
balancing_stack.append(")")
|
||||
elif data == "[":
|
||||
balancing_stack.append("]")
|
||||
elif data in ("}", ")", "]"):
|
||||
if not balancing_stack:
|
||||
raise TemplateSyntaxError(
|
||||
f"unexpected '{data}'", lineno, name, filename
|
||||
)
|
||||
|
||||
expected_op = balancing_stack.pop()
|
||||
|
||||
if expected_op != data:
|
||||
raise TemplateSyntaxError(
|
||||
f"unexpected '{data}', expected '{expected_op}'",
|
||||
lineno,
|
||||
name,
|
||||
filename,
|
||||
)
|
||||
|
||||
# yield items
|
||||
if data or tokens not in ignore_if_empty:
|
||||
yield lineno, tokens, data
|
||||
|
||||
lineno += data.count("\n")
|
||||
|
||||
line_starting = m.group()[-1:] == "\n"
|
||||
# fetch new position into new variable so that we can check
|
||||
# if there is a internal parsing error which would result
|
||||
# in an infinite loop
|
||||
pos2 = m.end()
|
||||
|
||||
# handle state changes
|
||||
if new_state is not None:
|
||||
# remove the uppermost state
|
||||
if new_state == "#pop":
|
||||
stack.pop()
|
||||
# resolve the new state by group checking
|
||||
elif new_state == "#bygroup":
|
||||
for key, value in m.groupdict().items():
|
||||
if value is not None:
|
||||
stack.append(key)
|
||||
break
|
||||
else:
|
||||
raise RuntimeError(
|
||||
f"{regex!r} wanted to resolve the new state dynamically"
|
||||
f" but no group matched"
|
||||
)
|
||||
# direct state name given
|
||||
else:
|
||||
stack.append(new_state)
|
||||
|
||||
statetokens = self.rules[stack[-1]]
|
||||
# we are still at the same position and no stack change.
|
||||
# this means a loop without break condition, avoid that and
|
||||
# raise error
|
||||
elif pos2 == pos:
|
||||
raise RuntimeError(
|
||||
f"{regex!r} yielded empty string without stack change"
|
||||
)
|
||||
|
||||
# publish new function and start again
|
||||
pos = pos2
|
||||
break
|
||||
# if loop terminated without break we haven't found a single match
|
||||
# either we are at the end of the file or we have a problem
|
||||
else:
|
||||
# end of text
|
||||
if pos >= source_length:
|
||||
return
|
||||
|
||||
# something went wrong
|
||||
raise TemplateSyntaxError(
|
||||
f"unexpected char {source[pos]!r} at {pos}", lineno, name, filename
|
||||
)
|
||||
693
venv/Lib/site-packages/jinja2/loaders.py
Normal file
693
venv/Lib/site-packages/jinja2/loaders.py
Normal file
@@ -0,0 +1,693 @@
|
||||
"""API and implementations for loading templates from different data
|
||||
sources.
|
||||
"""
|
||||
|
||||
import importlib.util
|
||||
import os
|
||||
import posixpath
|
||||
import sys
|
||||
import typing as t
|
||||
import weakref
|
||||
import zipimport
|
||||
from collections import abc
|
||||
from hashlib import sha1
|
||||
from importlib import import_module
|
||||
from types import ModuleType
|
||||
|
||||
from .exceptions import TemplateNotFound
|
||||
from .utils import internalcode
|
||||
|
||||
if t.TYPE_CHECKING:
|
||||
from .environment import Environment
|
||||
from .environment import Template
|
||||
|
||||
|
||||
def split_template_path(template: str) -> t.List[str]:
|
||||
"""Split a path into segments and perform a sanity check. If it detects
|
||||
'..' in the path it will raise a `TemplateNotFound` error.
|
||||
"""
|
||||
pieces = []
|
||||
for piece in template.split("/"):
|
||||
if (
|
||||
os.path.sep in piece
|
||||
or (os.path.altsep and os.path.altsep in piece)
|
||||
or piece == os.path.pardir
|
||||
):
|
||||
raise TemplateNotFound(template)
|
||||
elif piece and piece != ".":
|
||||
pieces.append(piece)
|
||||
return pieces
|
||||
|
||||
|
||||
class BaseLoader:
|
||||
"""Baseclass for all loaders. Subclass this and override `get_source` to
|
||||
implement a custom loading mechanism. The environment provides a
|
||||
`get_template` method that calls the loader's `load` method to get the
|
||||
:class:`Template` object.
|
||||
|
||||
A very basic example for a loader that looks up templates on the file
|
||||
system could look like this::
|
||||
|
||||
from jinja2 import BaseLoader, TemplateNotFound
|
||||
from os.path import join, exists, getmtime
|
||||
|
||||
class MyLoader(BaseLoader):
|
||||
|
||||
def __init__(self, path):
|
||||
self.path = path
|
||||
|
||||
def get_source(self, environment, template):
|
||||
path = join(self.path, template)
|
||||
if not exists(path):
|
||||
raise TemplateNotFound(template)
|
||||
mtime = getmtime(path)
|
||||
with open(path) as f:
|
||||
source = f.read()
|
||||
return source, path, lambda: mtime == getmtime(path)
|
||||
"""
|
||||
|
||||
#: if set to `False` it indicates that the loader cannot provide access
|
||||
#: to the source of templates.
|
||||
#:
|
||||
#: .. versionadded:: 2.4
|
||||
has_source_access = True
|
||||
|
||||
def get_source(
|
||||
self, environment: "Environment", template: str
|
||||
) -> t.Tuple[str, t.Optional[str], t.Optional[t.Callable[[], bool]]]:
|
||||
"""Get the template source, filename and reload helper for a template.
|
||||
It's passed the environment and template name and has to return a
|
||||
tuple in the form ``(source, filename, uptodate)`` or raise a
|
||||
`TemplateNotFound` error if it can't locate the template.
|
||||
|
||||
The source part of the returned tuple must be the source of the
|
||||
template as a string. The filename should be the name of the
|
||||
file on the filesystem if it was loaded from there, otherwise
|
||||
``None``. The filename is used by Python for the tracebacks
|
||||
if no loader extension is used.
|
||||
|
||||
The last item in the tuple is the `uptodate` function. If auto
|
||||
reloading is enabled it's always called to check if the template
|
||||
changed. No arguments are passed so the function must store the
|
||||
old state somewhere (for example in a closure). If it returns `False`
|
||||
the template will be reloaded.
|
||||
"""
|
||||
if not self.has_source_access:
|
||||
raise RuntimeError(
|
||||
f"{type(self).__name__} cannot provide access to the source"
|
||||
)
|
||||
raise TemplateNotFound(template)
|
||||
|
||||
def list_templates(self) -> t.List[str]:
|
||||
"""Iterates over all templates. If the loader does not support that
|
||||
it should raise a :exc:`TypeError` which is the default behavior.
|
||||
"""
|
||||
raise TypeError("this loader cannot iterate over all templates")
|
||||
|
||||
@internalcode
|
||||
def load(
|
||||
self,
|
||||
environment: "Environment",
|
||||
name: str,
|
||||
globals: t.Optional[t.MutableMapping[str, t.Any]] = None,
|
||||
) -> "Template":
|
||||
"""Loads a template. This method looks up the template in the cache
|
||||
or loads one by calling :meth:`get_source`. Subclasses should not
|
||||
override this method as loaders working on collections of other
|
||||
loaders (such as :class:`PrefixLoader` or :class:`ChoiceLoader`)
|
||||
will not call this method but `get_source` directly.
|
||||
"""
|
||||
code = None
|
||||
if globals is None:
|
||||
globals = {}
|
||||
|
||||
# first we try to get the source for this template together
|
||||
# with the filename and the uptodate function.
|
||||
source, filename, uptodate = self.get_source(environment, name)
|
||||
|
||||
# try to load the code from the bytecode cache if there is a
|
||||
# bytecode cache configured.
|
||||
bcc = environment.bytecode_cache
|
||||
if bcc is not None:
|
||||
bucket = bcc.get_bucket(environment, name, filename, source)
|
||||
code = bucket.code
|
||||
|
||||
# if we don't have code so far (not cached, no longer up to
|
||||
# date) etc. we compile the template
|
||||
if code is None:
|
||||
code = environment.compile(source, name, filename)
|
||||
|
||||
# if the bytecode cache is available and the bucket doesn't
|
||||
# have a code so far, we give the bucket the new code and put
|
||||
# it back to the bytecode cache.
|
||||
if bcc is not None and bucket.code is None:
|
||||
bucket.code = code
|
||||
bcc.set_bucket(bucket)
|
||||
|
||||
return environment.template_class.from_code(
|
||||
environment, code, globals, uptodate
|
||||
)
|
||||
|
||||
|
||||
class FileSystemLoader(BaseLoader):
|
||||
"""Load templates from a directory in the file system.
|
||||
|
||||
The path can be relative or absolute. Relative paths are relative to
|
||||
the current working directory.
|
||||
|
||||
.. code-block:: python
|
||||
|
||||
loader = FileSystemLoader("templates")
|
||||
|
||||
A list of paths can be given. The directories will be searched in
|
||||
order, stopping at the first matching template.
|
||||
|
||||
.. code-block:: python
|
||||
|
||||
loader = FileSystemLoader(["/override/templates", "/default/templates"])
|
||||
|
||||
:param searchpath: A path, or list of paths, to the directory that
|
||||
contains the templates.
|
||||
:param encoding: Use this encoding to read the text from template
|
||||
files.
|
||||
:param followlinks: Follow symbolic links in the path.
|
||||
|
||||
.. versionchanged:: 2.8
|
||||
Added the ``followlinks`` parameter.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
searchpath: t.Union[
|
||||
str, "os.PathLike[str]", t.Sequence[t.Union[str, "os.PathLike[str]"]]
|
||||
],
|
||||
encoding: str = "utf-8",
|
||||
followlinks: bool = False,
|
||||
) -> None:
|
||||
if not isinstance(searchpath, abc.Iterable) or isinstance(searchpath, str):
|
||||
searchpath = [searchpath]
|
||||
|
||||
self.searchpath = [os.fspath(p) for p in searchpath]
|
||||
self.encoding = encoding
|
||||
self.followlinks = followlinks
|
||||
|
||||
def get_source(
|
||||
self, environment: "Environment", template: str
|
||||
) -> t.Tuple[str, str, t.Callable[[], bool]]:
|
||||
pieces = split_template_path(template)
|
||||
|
||||
for searchpath in self.searchpath:
|
||||
# Use posixpath even on Windows to avoid "drive:" or UNC
|
||||
# segments breaking out of the search directory.
|
||||
filename = posixpath.join(searchpath, *pieces)
|
||||
|
||||
if os.path.isfile(filename):
|
||||
break
|
||||
else:
|
||||
plural = "path" if len(self.searchpath) == 1 else "paths"
|
||||
paths_str = ", ".join(repr(p) for p in self.searchpath)
|
||||
raise TemplateNotFound(
|
||||
template,
|
||||
f"{template!r} not found in search {plural}: {paths_str}",
|
||||
)
|
||||
|
||||
with open(filename, encoding=self.encoding) as f:
|
||||
contents = f.read()
|
||||
|
||||
mtime = os.path.getmtime(filename)
|
||||
|
||||
def uptodate() -> bool:
|
||||
try:
|
||||
return os.path.getmtime(filename) == mtime
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
# Use normpath to convert Windows altsep to sep.
|
||||
return contents, os.path.normpath(filename), uptodate
|
||||
|
||||
def list_templates(self) -> t.List[str]:
|
||||
found = set()
|
||||
for searchpath in self.searchpath:
|
||||
walk_dir = os.walk(searchpath, followlinks=self.followlinks)
|
||||
for dirpath, _, filenames in walk_dir:
|
||||
for filename in filenames:
|
||||
template = (
|
||||
os.path.join(dirpath, filename)[len(searchpath) :]
|
||||
.strip(os.path.sep)
|
||||
.replace(os.path.sep, "/")
|
||||
)
|
||||
if template[:2] == "./":
|
||||
template = template[2:]
|
||||
if template not in found:
|
||||
found.add(template)
|
||||
return sorted(found)
|
||||
|
||||
|
||||
if sys.version_info >= (3, 13):
|
||||
|
||||
def _get_zipimporter_files(z: t.Any) -> t.Dict[str, object]:
|
||||
try:
|
||||
get_files = z._get_files
|
||||
except AttributeError as e:
|
||||
raise TypeError(
|
||||
"This zip import does not have the required"
|
||||
" metadata to list templates."
|
||||
) from e
|
||||
return get_files()
|
||||
else:
|
||||
|
||||
def _get_zipimporter_files(z: t.Any) -> t.Dict[str, object]:
|
||||
try:
|
||||
files = z._files
|
||||
except AttributeError as e:
|
||||
raise TypeError(
|
||||
"This zip import does not have the required"
|
||||
" metadata to list templates."
|
||||
) from e
|
||||
return files # type: ignore[no-any-return]
|
||||
|
||||
|
||||
class PackageLoader(BaseLoader):
|
||||
"""Load templates from a directory in a Python package.
|
||||
|
||||
:param package_name: Import name of the package that contains the
|
||||
template directory.
|
||||
:param package_path: Directory within the imported package that
|
||||
contains the templates.
|
||||
:param encoding: Encoding of template files.
|
||||
|
||||
The following example looks up templates in the ``pages`` directory
|
||||
within the ``project.ui`` package.
|
||||
|
||||
.. code-block:: python
|
||||
|
||||
loader = PackageLoader("project.ui", "pages")
|
||||
|
||||
Only packages installed as directories (standard pip behavior) or
|
||||
zip/egg files (less common) are supported. The Python API for
|
||||
introspecting data in packages is too limited to support other
|
||||
installation methods the way this loader requires.
|
||||
|
||||
There is limited support for :pep:`420` namespace packages. The
|
||||
template directory is assumed to only be in one namespace
|
||||
contributor. Zip files contributing to a namespace are not
|
||||
supported.
|
||||
|
||||
.. versionchanged:: 3.0
|
||||
No longer uses ``setuptools`` as a dependency.
|
||||
|
||||
.. versionchanged:: 3.0
|
||||
Limited PEP 420 namespace package support.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
package_name: str,
|
||||
package_path: "str" = "templates",
|
||||
encoding: str = "utf-8",
|
||||
) -> None:
|
||||
package_path = os.path.normpath(package_path).rstrip(os.path.sep)
|
||||
|
||||
# normpath preserves ".", which isn't valid in zip paths.
|
||||
if package_path == os.path.curdir:
|
||||
package_path = ""
|
||||
elif package_path[:2] == os.path.curdir + os.path.sep:
|
||||
package_path = package_path[2:]
|
||||
|
||||
self.package_path = package_path
|
||||
self.package_name = package_name
|
||||
self.encoding = encoding
|
||||
|
||||
# Make sure the package exists. This also makes namespace
|
||||
# packages work, otherwise get_loader returns None.
|
||||
import_module(package_name)
|
||||
spec = importlib.util.find_spec(package_name)
|
||||
assert spec is not None, "An import spec was not found for the package."
|
||||
loader = spec.loader
|
||||
assert loader is not None, "A loader was not found for the package."
|
||||
self._loader = loader
|
||||
self._archive = None
|
||||
|
||||
if isinstance(loader, zipimport.zipimporter):
|
||||
self._archive = loader.archive
|
||||
pkgdir = next(iter(spec.submodule_search_locations)) # type: ignore
|
||||
template_root = os.path.join(pkgdir, package_path).rstrip(os.path.sep)
|
||||
else:
|
||||
roots: t.List[str] = []
|
||||
|
||||
# One element for regular packages, multiple for namespace
|
||||
# packages, or None for single module file.
|
||||
if spec.submodule_search_locations:
|
||||
roots.extend(spec.submodule_search_locations)
|
||||
# A single module file, use the parent directory instead.
|
||||
elif spec.origin is not None:
|
||||
roots.append(os.path.dirname(spec.origin))
|
||||
|
||||
if not roots:
|
||||
raise ValueError(
|
||||
f"The {package_name!r} package was not installed in a"
|
||||
" way that PackageLoader understands."
|
||||
)
|
||||
|
||||
for root in roots:
|
||||
root = os.path.join(root, package_path)
|
||||
|
||||
if os.path.isdir(root):
|
||||
template_root = root
|
||||
break
|
||||
else:
|
||||
raise ValueError(
|
||||
f"PackageLoader could not find a {package_path!r} directory"
|
||||
f" in the {package_name!r} package."
|
||||
)
|
||||
|
||||
self._template_root = template_root
|
||||
|
||||
def get_source(
|
||||
self, environment: "Environment", template: str
|
||||
) -> t.Tuple[str, str, t.Optional[t.Callable[[], bool]]]:
|
||||
# Use posixpath even on Windows to avoid "drive:" or UNC
|
||||
# segments breaking out of the search directory. Use normpath to
|
||||
# convert Windows altsep to sep.
|
||||
p = os.path.normpath(
|
||||
posixpath.join(self._template_root, *split_template_path(template))
|
||||
)
|
||||
up_to_date: t.Optional[t.Callable[[], bool]]
|
||||
|
||||
if self._archive is None:
|
||||
# Package is a directory.
|
||||
if not os.path.isfile(p):
|
||||
raise TemplateNotFound(template)
|
||||
|
||||
with open(p, "rb") as f:
|
||||
source = f.read()
|
||||
|
||||
mtime = os.path.getmtime(p)
|
||||
|
||||
def up_to_date() -> bool:
|
||||
return os.path.isfile(p) and os.path.getmtime(p) == mtime
|
||||
|
||||
else:
|
||||
# Package is a zip file.
|
||||
try:
|
||||
source = self._loader.get_data(p) # type: ignore
|
||||
except OSError as e:
|
||||
raise TemplateNotFound(template) from e
|
||||
|
||||
# Could use the zip's mtime for all template mtimes, but
|
||||
# would need to safely reload the module if it's out of
|
||||
# date, so just report it as always current.
|
||||
up_to_date = None
|
||||
|
||||
return source.decode(self.encoding), p, up_to_date
|
||||
|
||||
def list_templates(self) -> t.List[str]:
|
||||
results: t.List[str] = []
|
||||
|
||||
if self._archive is None:
|
||||
# Package is a directory.
|
||||
offset = len(self._template_root)
|
||||
|
||||
for dirpath, _, filenames in os.walk(self._template_root):
|
||||
dirpath = dirpath[offset:].lstrip(os.path.sep)
|
||||
results.extend(
|
||||
os.path.join(dirpath, name).replace(os.path.sep, "/")
|
||||
for name in filenames
|
||||
)
|
||||
else:
|
||||
files = _get_zipimporter_files(self._loader)
|
||||
|
||||
# Package is a zip file.
|
||||
prefix = (
|
||||
self._template_root[len(self._archive) :].lstrip(os.path.sep)
|
||||
+ os.path.sep
|
||||
)
|
||||
offset = len(prefix)
|
||||
|
||||
for name in files:
|
||||
# Find names under the templates directory that aren't directories.
|
||||
if name.startswith(prefix) and name[-1] != os.path.sep:
|
||||
results.append(name[offset:].replace(os.path.sep, "/"))
|
||||
|
||||
results.sort()
|
||||
return results
|
||||
|
||||
|
||||
class DictLoader(BaseLoader):
|
||||
"""Loads a template from a Python dict mapping template names to
|
||||
template source. This loader is useful for unittesting:
|
||||
|
||||
>>> loader = DictLoader({'index.html': 'source here'})
|
||||
|
||||
Because auto reloading is rarely useful this is disabled by default.
|
||||
"""
|
||||
|
||||
def __init__(self, mapping: t.Mapping[str, str]) -> None:
|
||||
self.mapping = mapping
|
||||
|
||||
def get_source(
|
||||
self, environment: "Environment", template: str
|
||||
) -> t.Tuple[str, None, t.Callable[[], bool]]:
|
||||
if template in self.mapping:
|
||||
source = self.mapping[template]
|
||||
return source, None, lambda: source == self.mapping.get(template)
|
||||
raise TemplateNotFound(template)
|
||||
|
||||
def list_templates(self) -> t.List[str]:
|
||||
return sorted(self.mapping)
|
||||
|
||||
|
||||
class FunctionLoader(BaseLoader):
|
||||
"""A loader that is passed a function which does the loading. The
|
||||
function receives the name of the template and has to return either
|
||||
a string with the template source, a tuple in the form ``(source,
|
||||
filename, uptodatefunc)`` or `None` if the template does not exist.
|
||||
|
||||
>>> def load_template(name):
|
||||
... if name == 'index.html':
|
||||
... return '...'
|
||||
...
|
||||
>>> loader = FunctionLoader(load_template)
|
||||
|
||||
The `uptodatefunc` is a function that is called if autoreload is enabled
|
||||
and has to return `True` if the template is still up to date. For more
|
||||
details have a look at :meth:`BaseLoader.get_source` which has the same
|
||||
return value.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
load_func: t.Callable[
|
||||
[str],
|
||||
t.Optional[
|
||||
t.Union[
|
||||
str, t.Tuple[str, t.Optional[str], t.Optional[t.Callable[[], bool]]]
|
||||
]
|
||||
],
|
||||
],
|
||||
) -> None:
|
||||
self.load_func = load_func
|
||||
|
||||
def get_source(
|
||||
self, environment: "Environment", template: str
|
||||
) -> t.Tuple[str, t.Optional[str], t.Optional[t.Callable[[], bool]]]:
|
||||
rv = self.load_func(template)
|
||||
|
||||
if rv is None:
|
||||
raise TemplateNotFound(template)
|
||||
|
||||
if isinstance(rv, str):
|
||||
return rv, None, None
|
||||
|
||||
return rv
|
||||
|
||||
|
||||
class PrefixLoader(BaseLoader):
|
||||
"""A loader that is passed a dict of loaders where each loader is bound
|
||||
to a prefix. The prefix is delimited from the template by a slash per
|
||||
default, which can be changed by setting the `delimiter` argument to
|
||||
something else::
|
||||
|
||||
loader = PrefixLoader({
|
||||
'app1': PackageLoader('mypackage.app1'),
|
||||
'app2': PackageLoader('mypackage.app2')
|
||||
})
|
||||
|
||||
By loading ``'app1/index.html'`` the file from the app1 package is loaded,
|
||||
by loading ``'app2/index.html'`` the file from the second.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self, mapping: t.Mapping[str, BaseLoader], delimiter: str = "/"
|
||||
) -> None:
|
||||
self.mapping = mapping
|
||||
self.delimiter = delimiter
|
||||
|
||||
def get_loader(self, template: str) -> t.Tuple[BaseLoader, str]:
|
||||
try:
|
||||
prefix, name = template.split(self.delimiter, 1)
|
||||
loader = self.mapping[prefix]
|
||||
except (ValueError, KeyError) as e:
|
||||
raise TemplateNotFound(template) from e
|
||||
return loader, name
|
||||
|
||||
def get_source(
|
||||
self, environment: "Environment", template: str
|
||||
) -> t.Tuple[str, t.Optional[str], t.Optional[t.Callable[[], bool]]]:
|
||||
loader, name = self.get_loader(template)
|
||||
try:
|
||||
return loader.get_source(environment, name)
|
||||
except TemplateNotFound as e:
|
||||
# re-raise the exception with the correct filename here.
|
||||
# (the one that includes the prefix)
|
||||
raise TemplateNotFound(template) from e
|
||||
|
||||
@internalcode
|
||||
def load(
|
||||
self,
|
||||
environment: "Environment",
|
||||
name: str,
|
||||
globals: t.Optional[t.MutableMapping[str, t.Any]] = None,
|
||||
) -> "Template":
|
||||
loader, local_name = self.get_loader(name)
|
||||
try:
|
||||
return loader.load(environment, local_name, globals)
|
||||
except TemplateNotFound as e:
|
||||
# re-raise the exception with the correct filename here.
|
||||
# (the one that includes the prefix)
|
||||
raise TemplateNotFound(name) from e
|
||||
|
||||
def list_templates(self) -> t.List[str]:
|
||||
result = []
|
||||
for prefix, loader in self.mapping.items():
|
||||
for template in loader.list_templates():
|
||||
result.append(prefix + self.delimiter + template)
|
||||
return result
|
||||
|
||||
|
||||
class ChoiceLoader(BaseLoader):
|
||||
"""This loader works like the `PrefixLoader` just that no prefix is
|
||||
specified. If a template could not be found by one loader the next one
|
||||
is tried.
|
||||
|
||||
>>> loader = ChoiceLoader([
|
||||
... FileSystemLoader('/path/to/user/templates'),
|
||||
... FileSystemLoader('/path/to/system/templates')
|
||||
... ])
|
||||
|
||||
This is useful if you want to allow users to override builtin templates
|
||||
from a different location.
|
||||
"""
|
||||
|
||||
def __init__(self, loaders: t.Sequence[BaseLoader]) -> None:
|
||||
self.loaders = loaders
|
||||
|
||||
def get_source(
|
||||
self, environment: "Environment", template: str
|
||||
) -> t.Tuple[str, t.Optional[str], t.Optional[t.Callable[[], bool]]]:
|
||||
for loader in self.loaders:
|
||||
try:
|
||||
return loader.get_source(environment, template)
|
||||
except TemplateNotFound:
|
||||
pass
|
||||
raise TemplateNotFound(template)
|
||||
|
||||
@internalcode
|
||||
def load(
|
||||
self,
|
||||
environment: "Environment",
|
||||
name: str,
|
||||
globals: t.Optional[t.MutableMapping[str, t.Any]] = None,
|
||||
) -> "Template":
|
||||
for loader in self.loaders:
|
||||
try:
|
||||
return loader.load(environment, name, globals)
|
||||
except TemplateNotFound:
|
||||
pass
|
||||
raise TemplateNotFound(name)
|
||||
|
||||
def list_templates(self) -> t.List[str]:
|
||||
found = set()
|
||||
for loader in self.loaders:
|
||||
found.update(loader.list_templates())
|
||||
return sorted(found)
|
||||
|
||||
|
||||
class _TemplateModule(ModuleType):
|
||||
"""Like a normal module but with support for weak references"""
|
||||
|
||||
|
||||
class ModuleLoader(BaseLoader):
|
||||
"""This loader loads templates from precompiled templates.
|
||||
|
||||
Example usage:
|
||||
|
||||
>>> loader = ModuleLoader('/path/to/compiled/templates')
|
||||
|
||||
Templates can be precompiled with :meth:`Environment.compile_templates`.
|
||||
"""
|
||||
|
||||
has_source_access = False
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
path: t.Union[
|
||||
str, "os.PathLike[str]", t.Sequence[t.Union[str, "os.PathLike[str]"]]
|
||||
],
|
||||
) -> None:
|
||||
package_name = f"_jinja2_module_templates_{id(self):x}"
|
||||
|
||||
# create a fake module that looks for the templates in the
|
||||
# path given.
|
||||
mod = _TemplateModule(package_name)
|
||||
|
||||
if not isinstance(path, abc.Iterable) or isinstance(path, str):
|
||||
path = [path]
|
||||
|
||||
mod.__path__ = [os.fspath(p) for p in path]
|
||||
|
||||
sys.modules[package_name] = weakref.proxy(
|
||||
mod, lambda x: sys.modules.pop(package_name, None)
|
||||
)
|
||||
|
||||
# the only strong reference, the sys.modules entry is weak
|
||||
# so that the garbage collector can remove it once the
|
||||
# loader that created it goes out of business.
|
||||
self.module = mod
|
||||
self.package_name = package_name
|
||||
|
||||
@staticmethod
|
||||
def get_template_key(name: str) -> str:
|
||||
return "tmpl_" + sha1(name.encode("utf-8")).hexdigest()
|
||||
|
||||
@staticmethod
|
||||
def get_module_filename(name: str) -> str:
|
||||
return ModuleLoader.get_template_key(name) + ".py"
|
||||
|
||||
@internalcode
|
||||
def load(
|
||||
self,
|
||||
environment: "Environment",
|
||||
name: str,
|
||||
globals: t.Optional[t.MutableMapping[str, t.Any]] = None,
|
||||
) -> "Template":
|
||||
key = self.get_template_key(name)
|
||||
module = f"{self.package_name}.{key}"
|
||||
mod = getattr(self.module, module, None)
|
||||
|
||||
if mod is None:
|
||||
try:
|
||||
mod = __import__(module, None, None, ["root"])
|
||||
except ImportError as e:
|
||||
raise TemplateNotFound(name) from e
|
||||
|
||||
# remove the entry from sys.modules, we only want the attribute
|
||||
# on the module object we have stored on the loader.
|
||||
sys.modules.pop(module, None)
|
||||
|
||||
if globals is None:
|
||||
globals = {}
|
||||
|
||||
return environment.template_class.from_module_dict(
|
||||
environment, mod.__dict__, globals
|
||||
)
|
||||
Reference in New Issue
Block a user