diff options
Diffstat (limited to 'deps/v8/third_party/jinja2/environment.py')
| -rw-r--r-- | deps/v8/third_party/jinja2/environment.py | 1095 |
1 files changed, 700 insertions, 395 deletions
diff --git a/deps/v8/third_party/jinja2/environment.py b/deps/v8/third_party/jinja2/environment.py index 8430390eea..ea04e8b443 100644 --- a/deps/v8/third_party/jinja2/environment.py +++ b/deps/v8/third_party/jinja2/environment.py @@ -1,25 +1,19 @@ -# -*- coding: utf-8 -*- """Classes for managing templates and their runtime and compile time options. """ import os -import sys +import typing +import typing as t import weakref +from collections import ChainMap +from functools import lru_cache from functools import partial from functools import reduce +from types import CodeType from markupsafe import Markup from . import nodes -from ._compat import encode_filename -from ._compat import implements_iterator -from ._compat import implements_to_string -from ._compat import iteritems -from ._compat import PY2 -from ._compat import PYPY -from ._compat import reraise -from ._compat import string_types -from ._compat import text_type from .compiler import CodeGenerator from .compiler import generate from .defaults import BLOCK_END_STRING @@ -44,25 +38,33 @@ from .exceptions import TemplatesNotFound from .exceptions import TemplateSyntaxError from .exceptions import UndefinedError from .lexer import get_lexer +from .lexer import Lexer from .lexer import TokenStream from .nodes import EvalContext from .parser import Parser from .runtime import Context from .runtime import new_context from .runtime import Undefined +from .utils import _PassArg from .utils import concat from .utils import consume -from .utils import have_async_gen from .utils import import_string from .utils import internalcode from .utils import LRUCache from .utils import missing -# for direct template usage we have up to ten living environments -_spontaneous_environments = LRUCache(10) +if t.TYPE_CHECKING: + import typing_extensions as te + from .bccache import BytecodeCache + from .ext import Extension + from .loaders import BaseLoader + +_env_bound = t.TypeVar("_env_bound", bound="Environment") -def get_spontaneous_environment(cls, *args): +# for direct template usage we have up to ten living environments +@lru_cache(maxsize=10) +def get_spontaneous_environment(cls: t.Type[_env_bound], *args: t.Any) -> _env_bound: """Return a new spontaneous environment. A spontaneous environment is used for templates created directly rather than through an existing environment. @@ -70,75 +72,74 @@ def get_spontaneous_environment(cls, *args): :param cls: Environment class to create. :param args: Positional arguments passed to environment. """ - key = (cls, args) + env = cls(*args) + env.shared = True + return env - try: - return _spontaneous_environments[key] - except KeyError: - _spontaneous_environments[key] = env = cls(*args) - env.shared = True - return env - -def create_cache(size): +def create_cache( + size: int, +) -> t.Optional[t.MutableMapping[t.Tuple[weakref.ref, str], "Template"]]: """Return the cache class for the given size.""" if size == 0: return None + if size < 0: return {} - return LRUCache(size) + return LRUCache(size) # type: ignore -def copy_cache(cache): + +def copy_cache( + cache: t.Optional[t.MutableMapping], +) -> t.Optional[t.MutableMapping[t.Tuple[weakref.ref, str], "Template"]]: """Create an empty copy of the given cache.""" if cache is None: return None - elif type(cache) is dict: + + if type(cache) is dict: return {} - return LRUCache(cache.capacity) + + return LRUCache(cache.capacity) # type: ignore -def load_extensions(environment, extensions): +def load_extensions( + environment: "Environment", + extensions: t.Sequence[t.Union[str, t.Type["Extension"]]], +) -> t.Dict[str, "Extension"]: """Load the extensions from the list and bind it to the environment. - Returns a dict of instantiated environments. + Returns a dict of instantiated extensions. """ result = {} + for extension in extensions: - if isinstance(extension, string_types): - extension = import_string(extension) - result[extension.identifier] = extension(environment) - return result + if isinstance(extension, str): + extension = t.cast(t.Type["Extension"], import_string(extension)) + result[extension.identifier] = extension(environment) -def fail_for_missing_callable(string, name): - msg = string % name - if isinstance(name, Undefined): - try: - name._fail_with_undefined_error() - except Exception as e: - msg = "%s (%s; did you forget to quote the callable name?)" % (msg, e) - raise TemplateRuntimeError(msg) + return result -def _environment_sanity_check(environment): +def _environment_config_check(environment: "Environment") -> "Environment": """Perform a sanity check on the environment.""" assert issubclass( environment.undefined, Undefined - ), "undefined must be a subclass of undefined because filters depend on it." + ), "'undefined' must be a subclass of 'jinja2.Undefined'." assert ( environment.block_start_string != environment.variable_start_string != environment.comment_start_string - ), "block, variable and comment start strings must be different" - assert environment.newline_sequence in ( + ), "block, variable and comment start strings must be different." + assert environment.newline_sequence in { "\r", "\r\n", "\n", - ), "newline_sequence set to unknown line ending string." + }, "'newline_sequence' must be one of '\\n', '\\r\\n', or '\\r'." return environment -class Environment(object): +class Environment: r"""The core component of Jinja is the `Environment`. It contains important shared variables like configuration, filters, tests, globals and others. Instances of this class may be modified if @@ -256,9 +257,8 @@ class Environment(object): See :ref:`bytecode-cache` for more information. `enable_async` - If set to true this enables async template execution which allows - you to take advantage of newer Python features. This requires - Python 3.6 or later. + If set to true this enables async template execution which + allows using async functions and generators. """ #: if this environment is sandboxed. Modifying this variable won't make @@ -271,7 +271,7 @@ class Environment(object): overlayed = False #: the environment this environment is linked to if it is an overlay - linked_to = None + linked_to: t.Optional["Environment"] = None #: shared environments have this set to `True`. A shared environment #: must not be modified @@ -279,36 +279,40 @@ class Environment(object): #: the class that is used for code generation. See #: :class:`~jinja2.compiler.CodeGenerator` for more information. - code_generator_class = CodeGenerator + code_generator_class: t.Type["CodeGenerator"] = CodeGenerator + + concat = "".join - #: the context class thatis used for templates. See + #: the context class that is used for templates. See #: :class:`~jinja2.runtime.Context` for more information. - context_class = Context + context_class: t.Type[Context] = Context + + template_class: t.Type["Template"] def __init__( self, - block_start_string=BLOCK_START_STRING, - block_end_string=BLOCK_END_STRING, - variable_start_string=VARIABLE_START_STRING, - variable_end_string=VARIABLE_END_STRING, - comment_start_string=COMMENT_START_STRING, - comment_end_string=COMMENT_END_STRING, - line_statement_prefix=LINE_STATEMENT_PREFIX, - line_comment_prefix=LINE_COMMENT_PREFIX, - trim_blocks=TRIM_BLOCKS, - lstrip_blocks=LSTRIP_BLOCKS, - newline_sequence=NEWLINE_SEQUENCE, - keep_trailing_newline=KEEP_TRAILING_NEWLINE, - extensions=(), - optimized=True, - undefined=Undefined, - finalize=None, - autoescape=False, - loader=None, - cache_size=400, - auto_reload=True, - bytecode_cache=None, - enable_async=False, + block_start_string: str = BLOCK_START_STRING, + block_end_string: str = BLOCK_END_STRING, + variable_start_string: str = VARIABLE_START_STRING, + variable_end_string: str = VARIABLE_END_STRING, + comment_start_string: str = COMMENT_START_STRING, + comment_end_string: str = COMMENT_END_STRING, + line_statement_prefix: t.Optional[str] = LINE_STATEMENT_PREFIX, + line_comment_prefix: t.Optional[str] = LINE_COMMENT_PREFIX, + trim_blocks: bool = TRIM_BLOCKS, + lstrip_blocks: bool = LSTRIP_BLOCKS, + newline_sequence: "te.Literal['\\n', '\\r\\n', '\\r']" = NEWLINE_SEQUENCE, + keep_trailing_newline: bool = KEEP_TRAILING_NEWLINE, + extensions: t.Sequence[t.Union[str, t.Type["Extension"]]] = (), + optimized: bool = True, + undefined: t.Type[Undefined] = Undefined, + finalize: t.Optional[t.Callable[..., t.Any]] = None, + autoescape: t.Union[bool, t.Callable[[t.Optional[str]], bool]] = False, + loader: t.Optional["BaseLoader"] = None, + cache_size: int = 400, + auto_reload: bool = True, + bytecode_cache: t.Optional["BytecodeCache"] = None, + enable_async: bool = False, ): # !!Important notice!! # The constructor accepts quite a few arguments that should be @@ -336,7 +340,7 @@ class Environment(object): self.keep_trailing_newline = keep_trailing_newline # runtime information - self.undefined = undefined + self.undefined: t.Type[Undefined] = undefined self.optimized = optimized self.finalize = finalize self.autoescape = autoescape @@ -358,52 +362,50 @@ class Environment(object): # load extensions self.extensions = load_extensions(self, extensions) - self.enable_async = enable_async - self.is_async = self.enable_async and have_async_gen - if self.is_async: - # runs patch_all() to enable async support - from . import asyncsupport # noqa: F401 - - _environment_sanity_check(self) + self.is_async = enable_async + _environment_config_check(self) - def add_extension(self, extension): + def add_extension(self, extension: t.Union[str, t.Type["Extension"]]) -> None: """Adds an extension after the environment was created. .. versionadded:: 2.5 """ self.extensions.update(load_extensions(self, [extension])) - def extend(self, **attributes): + def extend(self, **attributes: t.Any) -> None: """Add the items to the instance of the environment if they do not exist yet. This is used by :ref:`extensions <writing-extensions>` to register callbacks and configuration values without breaking inheritance. """ - for key, value in iteritems(attributes): + for key, value in attributes.items(): if not hasattr(self, key): setattr(self, key, value) def overlay( self, - block_start_string=missing, - block_end_string=missing, - variable_start_string=missing, - variable_end_string=missing, - comment_start_string=missing, - comment_end_string=missing, - line_statement_prefix=missing, - line_comment_prefix=missing, - trim_blocks=missing, - lstrip_blocks=missing, - extensions=missing, - optimized=missing, - undefined=missing, - finalize=missing, - autoescape=missing, - loader=missing, - cache_size=missing, - auto_reload=missing, - bytecode_cache=missing, - ): + block_start_string: str = missing, + block_end_string: str = missing, + variable_start_string: str = missing, + variable_end_string: str = missing, + comment_start_string: str = missing, + comment_end_string: str = missing, + line_statement_prefix: t.Optional[str] = missing, + line_comment_prefix: t.Optional[str] = missing, + trim_blocks: bool = missing, + lstrip_blocks: bool = missing, + newline_sequence: "te.Literal['\\n', '\\r\\n', '\\r']" = missing, + keep_trailing_newline: bool = missing, + extensions: t.Sequence[t.Union[str, t.Type["Extension"]]] = missing, + optimized: bool = missing, + undefined: t.Type[Undefined] = missing, + finalize: t.Optional[t.Callable[..., t.Any]] = missing, + autoescape: t.Union[bool, t.Callable[[t.Optional[str]], bool]] = missing, + loader: t.Optional["BaseLoader"] = missing, + cache_size: int = missing, + auto_reload: bool = missing, + bytecode_cache: t.Optional["BytecodeCache"] = missing, + enable_async: bool = False, + ) -> "Environment": """Create a new overlay environment that shares all the data with the current environment except for cache and the overridden attributes. Extensions cannot be removed for an overlayed environment. An overlayed @@ -414,16 +416,20 @@ class Environment(object): up completely. Not all attributes are truly linked, some are just copied over so modifications on the original environment may not shine through. + + .. versionchanged:: 3.1.2 + Added the ``newline_sequence``,, ``keep_trailing_newline``, + and ``enable_async`` parameters to match ``__init__``. """ args = dict(locals()) - del args["self"], args["cache_size"], args["extensions"] + del args["self"], args["cache_size"], args["extensions"], args["enable_async"] rv = object.__new__(self.__class__) rv.__dict__.update(self.__dict__) rv.overlayed = True rv.linked_to = self - for key, value in iteritems(args): + for key, value in args.items(): if value is not missing: setattr(rv, key, value) @@ -433,25 +439,33 @@ class Environment(object): rv.cache = copy_cache(self.cache) rv.extensions = {} - for key, value in iteritems(self.extensions): + for key, value in self.extensions.items(): rv.extensions[key] = value.bind(rv) if extensions is not missing: rv.extensions.update(load_extensions(rv, extensions)) - return _environment_sanity_check(rv) + if enable_async is not missing: + rv.is_async = enable_async - lexer = property(get_lexer, doc="The lexer for this environment.") + return _environment_config_check(rv) - def iter_extensions(self): + @property + def lexer(self) -> Lexer: + """The lexer for this environment.""" + return get_lexer(self) + + def iter_extensions(self) -> t.Iterator["Extension"]: """Iterates over the extensions by priority.""" return iter(sorted(self.extensions.values(), key=lambda x: x.priority)) - def getitem(self, obj, argument): + def getitem( + self, obj: t.Any, argument: t.Union[str, t.Any] + ) -> t.Union[t.Any, Undefined]: """Get an item or attribute of an object but prefer the item.""" try: return obj[argument] except (AttributeError, TypeError, LookupError): - if isinstance(argument, string_types): + if isinstance(argument, str): try: attr = str(argument) except Exception: @@ -463,9 +477,9 @@ class Environment(object): pass return self.undefined(obj=obj, name=argument) - def getattr(self, obj, attribute): + def getattr(self, obj: t.Any, attribute: str) -> t.Any: """Get an item or attribute of an object but prefer the attribute. - Unlike :meth:`getitem` the attribute *must* be a bytestring. + Unlike :meth:`getitem` the attribute *must* be a string. """ try: return getattr(obj, attribute) @@ -476,51 +490,113 @@ class Environment(object): except (TypeError, LookupError, AttributeError): return self.undefined(obj=obj, name=attribute) - def call_filter( - self, name, value, args=None, kwargs=None, context=None, eval_ctx=None - ): - """Invokes a filter on a value the same way the compiler does it. + def _filter_test_common( + self, + name: t.Union[str, Undefined], + value: t.Any, + args: t.Optional[t.Sequence[t.Any]], + kwargs: t.Optional[t.Mapping[str, t.Any]], + context: t.Optional[Context], + eval_ctx: t.Optional[EvalContext], + is_filter: bool, + ) -> t.Any: + if is_filter: + env_map = self.filters + type_name = "filter" + else: + env_map = self.tests + type_name = "test" - Note that on Python 3 this might return a coroutine in case the - filter is running from an environment in async mode and the filter - supports async execution. It's your responsibility to await this - if needed. + func = env_map.get(name) # type: ignore - .. versionadded:: 2.7 - """ - func = self.filters.get(name) if func is None: - fail_for_missing_callable("no filter named %r", name) - args = [value] + list(args or ()) - if getattr(func, "contextfilter", False) is True: + msg = f"No {type_name} named {name!r}." + + if isinstance(name, Undefined): + try: + name._fail_with_undefined_error() + except Exception as e: + msg = f"{msg} ({e}; did you forget to quote the callable name?)" + + raise TemplateRuntimeError(msg) + + args = [value, *(args if args is not None else ())] + kwargs = kwargs if kwargs is not None else {} + pass_arg = _PassArg.from_obj(func) + + if pass_arg is _PassArg.context: if context is None: raise TemplateRuntimeError( - "Attempted to invoke context filter without context" + f"Attempted to invoke a context {type_name} without context." ) + args.insert(0, context) - elif getattr(func, "evalcontextfilter", False) is True: + elif pass_arg is _PassArg.eval_context: if eval_ctx is None: if context is not None: eval_ctx = context.eval_ctx else: eval_ctx = EvalContext(self) + args.insert(0, eval_ctx) - elif getattr(func, "environmentfilter", False) is True: + elif pass_arg is _PassArg.environment: args.insert(0, self) - return func(*args, **(kwargs or {})) - def call_test(self, name, value, args=None, kwargs=None): - """Invokes a test on a value the same way the compiler does it. + return func(*args, **kwargs) + + def call_filter( + self, + name: str, + value: t.Any, + args: t.Optional[t.Sequence[t.Any]] = None, + kwargs: t.Optional[t.Mapping[str, t.Any]] = None, + context: t.Optional[Context] = None, + eval_ctx: t.Optional[EvalContext] = None, + ) -> t.Any: + """Invoke a filter on a value the same way the compiler does. + + This might return a coroutine if the filter is running from an + environment in async mode and the filter supports async + execution. It's your responsibility to await this if needed. + + .. versionadded:: 2.7 + """ + return self._filter_test_common( + name, value, args, kwargs, context, eval_ctx, True + ) + + def call_test( + self, + name: str, + value: t.Any, + args: t.Optional[t.Sequence[t.Any]] = None, + kwargs: t.Optional[t.Mapping[str, t.Any]] = None, + context: t.Optional[Context] = None, + eval_ctx: t.Optional[EvalContext] = None, + ) -> t.Any: + """Invoke a test on a value the same way the compiler does. + + This might return a coroutine if the test is running from an + environment in async mode and the test supports async execution. + It's your responsibility to await this if needed. + + .. versionchanged:: 3.0 + Tests support ``@pass_context``, etc. decorators. Added + the ``context`` and ``eval_ctx`` parameters. .. versionadded:: 2.7 """ - func = self.tests.get(name) - if func is None: - fail_for_missing_callable("no test named %r", name) - return func(value, *(args or ()), **(kwargs or {})) + return self._filter_test_common( + name, value, args, kwargs, context, eval_ctx, False + ) @internalcode - def parse(self, source, name=None, filename=None): + def parse( + self, + source: str, + name: t.Optional[str] = None, + filename: t.Optional[str] = None, + ) -> nodes.Template: """Parse the sourcecode and return the abstract syntax tree. This tree of nodes is used by the compiler to convert the template into executable source- or bytecode. This is useful for debugging or to @@ -534,11 +610,18 @@ class Environment(object): except TemplateSyntaxError: self.handle_exception(source=source) - def _parse(self, source, name, filename): + def _parse( + self, source: str, name: t.Optional[str], filename: t.Optional[str] + ) -> nodes.Template: """Internal parsing function used by `parse` and `compile`.""" - return Parser(self, source, name, encode_filename(filename)).parse() + return Parser(self, source, name, filename).parse() - def lex(self, source, name=None, filename=None): + def lex( + self, + source: str, + name: t.Optional[str] = None, + filename: t.Optional[str] = None, + ) -> t.Iterator[t.Tuple[int, str, str]]: """Lex the given sourcecode and return a generator that yields tokens as tuples in the form ``(lineno, token_type, value)``. This can be useful for :ref:`extension development <writing-extensions>` @@ -548,13 +631,18 @@ class Environment(object): of the extensions to be applied you have to filter source through the :meth:`preprocess` method. """ - source = text_type(source) + source = str(source) try: return self.lexer.tokeniter(source, name, filename) except TemplateSyntaxError: self.handle_exception(source=source) - def preprocess(self, source, name=None, filename=None): + def preprocess( + self, + source: str, + name: t.Optional[str] = None, + filename: t.Optional[str] = None, + ) -> str: """Preprocesses the source with all extensions. This is automatically called for all parsing and compiling methods but *not* for :meth:`lex` because there you usually only want the actual source tokenized. @@ -562,28 +650,43 @@ class Environment(object): return reduce( lambda s, e: e.preprocess(s, name, filename), self.iter_extensions(), - text_type(source), + str(source), ) - def _tokenize(self, source, name, filename=None, state=None): + def _tokenize( + self, + source: str, + name: t.Optional[str], + filename: t.Optional[str] = None, + state: t.Optional[str] = None, + ) -> TokenStream: """Called by the parser to do the preprocessing and filtering for all the extensions. Returns a :class:`~jinja2.lexer.TokenStream`. """ source = self.preprocess(source, name, filename) stream = self.lexer.tokenize(source, name, filename, state) + for ext in self.iter_extensions(): - stream = ext.filter_stream(stream) + stream = ext.filter_stream(stream) # type: ignore + if not isinstance(stream, TokenStream): - stream = TokenStream(stream, name, filename) + stream = TokenStream(stream, name, filename) # type: ignore + return stream - def _generate(self, source, name, filename, defer_init=False): + def _generate( + self, + source: nodes.Template, + name: t.Optional[str], + filename: t.Optional[str], + defer_init: bool = False, + ) -> str: """Internal hook that can be overridden to hook a different generate method in. .. versionadded:: 2.5 """ - return generate( + return generate( # type: ignore source, self, name, @@ -592,16 +695,45 @@ class Environment(object): optimized=self.optimized, ) - def _compile(self, source, filename): + def _compile(self, source: str, filename: str) -> CodeType: """Internal hook that can be overridden to hook a different compile method in. .. versionadded:: 2.5 """ - return compile(source, filename, "exec") + return compile(source, filename, "exec") # type: ignore + + @typing.overload + def compile( # type: ignore + self, + source: t.Union[str, nodes.Template], + name: t.Optional[str] = None, + filename: t.Optional[str] = None, + raw: "te.Literal[False]" = False, + defer_init: bool = False, + ) -> CodeType: + ... + + @typing.overload + def compile( + self, + source: t.Union[str, nodes.Template], + name: t.Optional[str] = None, + filename: t.Optional[str] = None, + raw: "te.Literal[True]" = ..., + defer_init: bool = False, + ) -> str: + ... @internalcode - def compile(self, source, name=None, filename=None, raw=False, defer_init=False): + def compile( + self, + source: t.Union[str, nodes.Template], + name: t.Optional[str] = None, + filename: t.Optional[str] = None, + raw: bool = False, + defer_init: bool = False, + ) -> t.Union[str, CodeType]: """Compile a node or template source code. The `name` parameter is the load name of the template after it was joined using :meth:`join_path` if necessary, not the filename on the file system. @@ -623,7 +755,7 @@ class Environment(object): """ source_hint = None try: - if isinstance(source, string_types): + if isinstance(source, str): source_hint = source source = self._parse(source, name, filename) source = self._generate(source, name, filename, defer_init=defer_init) @@ -631,13 +763,13 @@ class Environment(object): return source if filename is None: filename = "<template>" - else: - filename = encode_filename(filename) return self._compile(source, filename) except TemplateSyntaxError: self.handle_exception(source=source_hint) - def compile_expression(self, source, undefined_to_none=True): + def compile_expression( + self, source: str, undefined_to_none: bool = True + ) -> "TemplateExpression": """A handy helper method that returns a callable that accepts keyword arguments that appear as variables in the expression. If called it returns the result of the expression. @@ -674,8 +806,7 @@ class Environment(object): ) expr.set_environment(self) except TemplateSyntaxError: - if sys.exc_info() is not None: - self.handle_exception(source=source) + self.handle_exception(source=source) body = [nodes.Assign(nodes.Name("result", "store"), expr, lineno=1)] template = self.from_string(nodes.Template(body, lineno=1)) @@ -683,14 +814,13 @@ class Environment(object): def compile_templates( self, - target, - extensions=None, - filter_func=None, - zip="deflated", - log_function=None, - ignore_errors=True, - py_compile=False, - ): + target: t.Union[str, os.PathLike], + extensions: t.Optional[t.Collection[str]] = None, + filter_func: t.Optional[t.Callable[[str], bool]] = None, + zip: t.Optional[str] = "deflated", + log_function: t.Optional[t.Callable[[str], None]] = None, + ignore_errors: bool = True, + ) -> None: """Finds all the templates the loader can find, compiles them and stores them in `target`. If `zip` is `None`, instead of in a zipfile, the templates will be stored in a directory. @@ -706,52 +836,26 @@ class Environment(object): syntax errors to abort the compilation you can set `ignore_errors` to `False` and you will get an exception on syntax errors. - If `py_compile` is set to `True` .pyc files will be written to the - target instead of standard .py files. This flag does not do anything - on pypy and Python 3 where pyc files are not picked up by itself and - don't give much benefit. - .. versionadded:: 2.4 """ from .loaders import ModuleLoader if log_function is None: - def log_function(x): + def log_function(x: str) -> None: pass - if py_compile: - if not PY2 or PYPY: - import warnings - - warnings.warn( - "'py_compile=True' has no effect on PyPy or Python" - " 3 and will be removed in version 3.0", - DeprecationWarning, - stacklevel=2, - ) - py_compile = False - else: - import imp - import marshal - - py_header = imp.get_magic() + u"\xff\xff\xff\xff".encode("iso-8859-15") + assert log_function is not None + assert self.loader is not None, "No loader configured." - # Python 3.3 added a source filesize to the header - if sys.version_info >= (3, 3): - py_header += u"\x00\x00\x00\x00".encode("iso-8859-15") - - def write_file(filename, data): + def write_file(filename: str, data: str) -> None: if zip: info = ZipInfo(filename) info.external_attr = 0o755 << 16 zip_file.writestr(info, data) else: - if isinstance(data, text_type): - data = data.encode("utf8") - with open(os.path.join(target, filename), "wb") as f: - f.write(data) + f.write(data.encode("utf8")) if zip is not None: from zipfile import ZipFile, ZipInfo, ZIP_DEFLATED, ZIP_STORED @@ -759,11 +863,11 @@ class Environment(object): zip_file = ZipFile( target, "w", dict(deflated=ZIP_DEFLATED, stored=ZIP_STORED)[zip] ) - log_function('Compiling into Zip archive "%s"' % target) + log_function(f"Compiling into Zip archive {target!r}") else: if not os.path.isdir(target): os.makedirs(target) - log_function('Compiling into folder "%s"' % target) + log_function(f"Compiling into folder {target!r}") try: for name in self.list_templates(extensions, filter_func): @@ -773,25 +877,24 @@ class Environment(object): except TemplateSyntaxError as e: if not ignore_errors: raise - log_function('Could not compile "%s": %s' % (name, e)) + log_function(f'Could not compile "{name}": {e}') continue filename = ModuleLoader.get_module_filename(name) - if py_compile: - c = self._compile(code, encode_filename(filename)) - write_file(filename + "c", py_header + marshal.dumps(c)) - log_function('Byte-compiled "%s" as %s' % (name, filename + "c")) - else: - write_file(filename, code) - log_function('Compiled "%s" as %s' % (name, filename)) + write_file(filename, code) + log_function(f'Compiled "{name}" as {filename}') finally: if zip: zip_file.close() log_function("Finished compiling templates") - def list_templates(self, extensions=None, filter_func=None): + def list_templates( + self, + extensions: t.Optional[t.Collection[str]] = None, + filter_func: t.Optional[t.Callable[[str], bool]] = None, + ) -> t.List[str]: """Returns a list of templates for this environment. This requires that the loader supports the loader's :meth:`~BaseLoader.list_templates` method. @@ -807,6 +910,7 @@ class Environment(object): .. versionadded:: 2.4 """ + assert self.loader is not None, "No loader configured." names = self.loader.list_templates() if extensions is not None: @@ -815,23 +919,23 @@ class Environment(object): "either extensions or filter_func can be passed, but not both" ) - def filter_func(x): - return "." in x and x.rsplit(".", 1)[1] in extensions + def filter_func(x: str) -> bool: + return "." in x and x.rsplit(".", 1)[1] in extensions # type: ignore if filter_func is not None: names = [name for name in names if filter_func(name)] return names - def handle_exception(self, source=None): + def handle_exception(self, source: t.Optional[str] = None) -> "te.NoReturn": """Exception handling helper. This is used internally to either raise rewritten exceptions or return a rendered traceback for the template. """ from .debug import rewrite_traceback_stack - reraise(*rewrite_traceback_stack(source=source)) + raise rewrite_traceback_stack(source=source) - def join_path(self, template, parent): + def join_path(self, template: str, parent: str) -> str: """Join a template with the parent. By default all the lookups are relative to the loader root so this method returns the `template` parameter unchanged, but if the paths should be relative to the @@ -844,7 +948,9 @@ class Environment(object): return template @internalcode - def _load_template(self, name, globals): + def _load_template( + self, name: str, globals: t.Optional[t.MutableMapping[str, t.Any]] + ) -> "Template": if self.loader is None: raise TypeError("no loader for this environment specified") cache_key = (weakref.ref(self.loader), name) @@ -853,49 +959,88 @@ class Environment(object): if template is not None and ( not self.auto_reload or template.is_up_to_date ): + # template.globals is a ChainMap, modifying it will only + # affect the template, not the environment globals. + if globals: + template.globals.update(globals) + return template - template = self.loader.load(self, name, globals) + + template = self.loader.load(self, name, self.make_globals(globals)) + if self.cache is not None: self.cache[cache_key] = template return template @internalcode - def get_template(self, name, parent=None, globals=None): - """Load a template from the loader. If a loader is configured this - method asks the loader for the template and returns a :class:`Template`. - If the `parent` parameter is not `None`, :meth:`join_path` is called - to get the real template name before loading. - - The `globals` parameter can be used to provide template wide globals. - These variables are available in the context at render time. - - If the template does not exist a :exc:`TemplateNotFound` exception is - raised. + def get_template( + self, + name: t.Union[str, "Template"], + parent: t.Optional[str] = None, + globals: t.Optional[t.MutableMapping[str, t.Any]] = None, + ) -> "Template": + """Load a template by name with :attr:`loader` and return a + :class:`Template`. If the template does not exist a + :exc:`TemplateNotFound` exception is raised. + + :param name: Name of the template to load. When loading + templates from the filesystem, "/" is used as the path + separator, even on Windows. + :param parent: The name of the parent template importing this + template. :meth:`join_path` can be used to implement name + transformations with this. + :param globals: Extend the environment :attr:`globals` with + these extra variables available for all renders of this + template. If the template has already been loaded and + cached, its globals are updated with any new items. + + .. versionchanged:: 3.0 + If a template is loaded from cache, ``globals`` will update + the template's globals instead of ignoring the new values. .. versionchanged:: 2.4 - If `name` is a :class:`Template` object it is returned from the - function unchanged. + If ``name`` is a :class:`Template` object it is returned + unchanged. """ if isinstance(name, Template): return name if parent is not None: name = self.join_path(name, parent) - return self._load_template(name, self.make_globals(globals)) + + return self._load_template(name, globals) @internalcode - def select_template(self, names, parent=None, globals=None): - """Works like :meth:`get_template` but tries a number of templates - before it fails. If it cannot find any of the templates, it will - raise a :exc:`TemplatesNotFound` exception. + def select_template( + self, + names: t.Iterable[t.Union[str, "Template"]], + parent: t.Optional[str] = None, + globals: t.Optional[t.MutableMapping[str, t.Any]] = None, + ) -> "Template": + """Like :meth:`get_template`, but tries loading multiple names. + If none of the names can be loaded a :exc:`TemplatesNotFound` + exception is raised. + + :param names: List of template names to try loading in order. + :param parent: The name of the parent template importing this + template. :meth:`join_path` can be used to implement name + transformations with this. + :param globals: Extend the environment :attr:`globals` with + these extra variables available for all renders of this + template. If the template has already been loaded and + cached, its globals are updated with any new items. + + .. versionchanged:: 3.0 + If a template is loaded from cache, ``globals`` will update + the template's globals instead of ignoring the new values. .. versionchanged:: 2.11 - If names is :class:`Undefined`, an :exc:`UndefinedError` is - raised instead. If no templates were found and names + If ``names`` is :class:`Undefined`, an :exc:`UndefinedError` + is raised instead. If no templates were found and ``names`` contains :class:`Undefined`, the message is more helpful. .. versionchanged:: 2.4 - If `names` contains a :class:`Template` object it is returned - from the function unchanged. + If ``names`` contains a :class:`Template` object it is + returned unchanged. .. versionadded:: 2.3 """ @@ -904,9 +1049,9 @@ class Environment(object): if not names: raise TemplatesNotFound( - message=u"Tried to select from an empty list " u"of templates." + message="Tried to select from an empty list of templates." ) - globals = self.make_globals(globals) + for name in names: if isinstance(name, Template): return name @@ -916,95 +1061,127 @@ class Environment(object): return self._load_template(name, globals) except (TemplateNotFound, UndefinedError): pass - raise TemplatesNotFound(names) + raise TemplatesNotFound(names) # type: ignore @internalcode - def get_or_select_template(self, template_name_or_list, parent=None, globals=None): - """Does a typecheck and dispatches to :meth:`select_template` - if an iterable of template names is given, otherwise to - :meth:`get_template`. + def get_or_select_template( + self, + template_name_or_list: t.Union[ + str, "Template", t.List[t.Union[str, "Template"]] + ], + parent: t.Optional[str] = None, + globals: t.Optional[t.MutableMapping[str, t.Any]] = None, + ) -> "Template": + """Use :meth:`select_template` if an iterable of template names + is given, or :meth:`get_template` if one name is given. .. versionadded:: 2.3 """ - if isinstance(template_name_or_list, (string_types, Undefined)): + if isinstance(template_name_or_list, (str, Undefined)): return self.get_template(template_name_or_list, parent, globals) elif isinstance(template_name_or_list, Template): return template_name_or_list return self.select_template(template_name_or_list, parent, globals) - def from_string(self, source, globals=None, template_class=None): - """Load a template from a string. This parses the source given and - returns a :class:`Template` object. + def from_string( + self, + source: t.Union[str, nodes.Template], + globals: t.Optional[t.MutableMapping[str, t.Any]] = None, + template_class: t.Optional[t.Type["Template"]] = None, + ) -> "Template": + """Load a template from a source string without using + :attr:`loader`. + + :param source: Jinja source to compile into a template. + :param globals: Extend the environment :attr:`globals` with + these extra variables available for all renders of this + template. If the template has already been loaded and + cached, its globals are updated with any new items. + :param template_class: Return an instance of this + :class:`Template` class. """ - globals = self.make_globals(globals) + gs = self.make_globals(globals) cls = template_class or self.template_class - return cls.from_code(self, self.compile(source), globals, None) - - def make_globals(self, d): - """Return a dict for the globals.""" - if not d: - return self.globals - return dict(self.globals, **d) - - -class Template(object): - """The central template object. This class represents a compiled template - and is used to evaluate it. - - Normally the template object is generated from an :class:`Environment` but - it also has a constructor that makes it possible to create a template - instance directly using the constructor. It takes the same arguments as - the environment constructor but it's not possible to specify a loader. - - Every template object has a few methods and members that are guaranteed - to exist. However it's important that a template object should be - considered immutable. Modifications on the object are not supported. - - Template objects created from the constructor rather than an environment - do have an `environment` attribute that points to a temporary environment - that is probably shared with other templates created with the constructor - and compatible settings. - - >>> template = Template('Hello {{ name }}!') - >>> template.render(name='John Doe') == u'Hello John Doe!' - True - >>> stream = template.stream(name='John Doe') - >>> next(stream) == u'Hello John Doe!' - True - >>> next(stream) - Traceback (most recent call last): - ... - StopIteration + return cls.from_code(self, self.compile(source), gs, None) + + def make_globals( + self, d: t.Optional[t.MutableMapping[str, t.Any]] + ) -> t.MutableMapping[str, t.Any]: + """Make the globals map for a template. Any given template + globals overlay the environment :attr:`globals`. + + Returns a :class:`collections.ChainMap`. This allows any changes + to a template's globals to only affect that template, while + changes to the environment's globals are still reflected. + However, avoid modifying any globals after a template is loaded. + + :param d: Dict of template-specific globals. + + .. versionchanged:: 3.0 + Use :class:`collections.ChainMap` to always prevent mutating + environment globals. + """ + if d is None: + d = {} + + return ChainMap(d, self.globals) + + +class Template: + """A compiled template that can be rendered. + + Use the methods on :class:`Environment` to create or load templates. + The environment is used to configure how templates are compiled and + behave. + + It is also possible to create a template object directly. This is + not usually recommended. The constructor takes most of the same + arguments as :class:`Environment`. All templates created with the + same environment arguments share the same ephemeral ``Environment`` + instance behind the scenes. + + A template object should be considered immutable. Modifications on + the object are not supported. """ #: Type of environment to create when creating a template directly #: rather than through an existing environment. - environment_class = Environment + environment_class: t.Type[Environment] = Environment + + environment: Environment + globals: t.MutableMapping[str, t.Any] + name: t.Optional[str] + filename: t.Optional[str] + blocks: t.Dict[str, t.Callable[[Context], t.Iterator[str]]] + root_render_func: t.Callable[[Context], t.Iterator[str]] + _module: t.Optional["TemplateModule"] + _debug_info: str + _uptodate: t.Optional[t.Callable[[], bool]] def __new__( cls, - source, - block_start_string=BLOCK_START_STRING, - block_end_string=BLOCK_END_STRING, - variable_start_string=VARIABLE_START_STRING, - variable_end_string=VARIABLE_END_STRING, - comment_start_string=COMMENT_START_STRING, - comment_end_string=COMMENT_END_STRING, - line_statement_prefix=LINE_STATEMENT_PREFIX, - line_comment_prefix=LINE_COMMENT_PREFIX, - trim_blocks=TRIM_BLOCKS, - lstrip_blocks=LSTRIP_BLOCKS, - newline_sequence=NEWLINE_SEQUENCE, - keep_trailing_newline=KEEP_TRAILING_NEWLINE, - extensions=(), - optimized=True, - undefined=Undefined, - finalize=None, - autoescape=False, - enable_async=False, - ): + source: t.Union[str, nodes.Template], + block_start_string: str = BLOCK_START_STRING, + block_end_string: str = BLOCK_END_STRING, + variable_start_string: str = VARIABLE_START_STRING, + variable_end_string: str = VARIABLE_END_STRING, + comment_start_string: str = COMMENT_START_STRING, + comment_end_string: str = COMMENT_END_STRING, + line_statement_prefix: t.Optional[str] = LINE_STATEMENT_PREFIX, + line_comment_prefix: t.Optional[str] = LINE_COMMENT_PREFIX, + trim_blocks: bool = TRIM_BLOCKS, + lstrip_blocks: bool = LSTRIP_BLOCKS, + newline_sequence: "te.Literal['\\n', '\\r\\n', '\\r']" = NEWLINE_SEQUENCE, + keep_trailing_newline: bool = KEEP_TRAILING_NEWLINE, + extensions: t.Sequence[t.Union[str, t.Type["Extension"]]] = (), + optimized: bool = True, + undefined: t.Type[Undefined] = Undefined, + finalize: t.Optional[t.Callable[..., t.Any]] = None, + autoescape: t.Union[bool, t.Callable[[t.Optional[str]], bool]] = False, + enable_async: bool = False, + ) -> t.Any: # it returns a `Template`, but this breaks the sphinx build... env = get_spontaneous_environment( - cls.environment_class, + cls.environment_class, # type: ignore block_start_string, block_end_string, variable_start_string, @@ -1019,7 +1196,7 @@ class Template(object): keep_trailing_newline, frozenset(extensions), optimized, - undefined, + undefined, # type: ignore finalize, autoescape, None, @@ -1031,7 +1208,13 @@ class Template(object): return env.from_string(source, template_class=cls) @classmethod - def from_code(cls, environment, code, globals, uptodate=None): + def from_code( + cls, + environment: Environment, + code: CodeType, + globals: t.MutableMapping[str, t.Any], + uptodate: t.Optional[t.Callable[[], bool]] = None, + ) -> "Template": """Creates a template object from compiled code and the globals. This is used by the loaders and environment to create a template object. """ @@ -1042,7 +1225,12 @@ class Template(object): return rv @classmethod - def from_module_dict(cls, environment, module_dict, globals): + def from_module_dict( + cls, + environment: Environment, + module_dict: t.MutableMapping[str, t.Any], + globals: t.MutableMapping[str, t.Any], + ) -> "Template": """Creates a template object from a module. This is used by the module loader to create a template object. @@ -1051,8 +1239,13 @@ class Template(object): return cls._from_namespace(environment, module_dict, globals) @classmethod - def _from_namespace(cls, environment, namespace, globals): - t = object.__new__(cls) + def _from_namespace( + cls, + environment: Environment, + namespace: t.MutableMapping[str, t.Any], + globals: t.MutableMapping[str, t.Any], + ) -> "Template": + t: "Template" = object.__new__(cls) t.environment = environment t.globals = globals t.name = namespace["name"] @@ -1060,7 +1253,7 @@ class Template(object): t.blocks = namespace["blocks"] # render function and module - t.root_render_func = namespace["root"] + t.root_render_func = namespace["root"] # type: ignore t._module = None # debug and loader helpers @@ -1073,7 +1266,7 @@ class Template(object): return t - def render(self, *args, **kwargs): + def render(self, *args: t.Any, **kwargs: t.Any) -> str: """This method accepts the same arguments as the `dict` constructor: A dict, a dict subclass or some keyword arguments. If no arguments are given the context will be empty. These two calls do the same:: @@ -1081,15 +1274,33 @@ class Template(object): template.render(knights='that say nih') template.render({'knights': 'that say nih'}) - This will return the rendered template as unicode string. + This will return the rendered template as a string. """ - vars = dict(*args, **kwargs) + if self.environment.is_async: + import asyncio + + close = False + + try: + loop = asyncio.get_running_loop() + except RuntimeError: + loop = asyncio.new_event_loop() + close = True + + try: + return loop.run_until_complete(self.render_async(*args, **kwargs)) + finally: + if close: + loop.close() + + ctx = self.new_context(dict(*args, **kwargs)) + try: - return concat(self.root_render_func(self.new_context(vars))) + return self.environment.concat(self.root_render_func(ctx)) # type: ignore except Exception: self.environment.handle_exception() - def render_async(self, *args, **kwargs): + async def render_async(self, *args: t.Any, **kwargs: t.Any) -> str: """This works similar to :meth:`render` but returns a coroutine that when awaited returns the entire rendered template string. This requires the async feature to be enabled. @@ -1098,42 +1309,75 @@ class Template(object): await template.render_async(knights='that say nih; asynchronously') """ - # see asyncsupport for the actual implementation - raise NotImplementedError( - "This feature is not available for this version of Python" - ) + if not self.environment.is_async: + raise RuntimeError( + "The environment was not created with async mode enabled." + ) + + ctx = self.new_context(dict(*args, **kwargs)) + + try: + return self.environment.concat( # type: ignore + [n async for n in self.root_render_func(ctx)] # type: ignore + ) + except Exception: + return self.environment.handle_exception() - def stream(self, *args, **kwargs): + def stream(self, *args: t.Any, **kwargs: t.Any) -> "TemplateStream": """Works exactly like :meth:`generate` but returns a :class:`TemplateStream`. """ return TemplateStream(self.generate(*args, **kwargs)) - def generate(self, *args, **kwargs): + def generate(self, *args: t.Any, **kwargs: t.Any) -> t.Iterator[str]: """For very large templates it can be useful to not render the whole template at once but evaluate each statement after another and yield piece for piece. This method basically does exactly that and returns - a generator that yields one item after another as unicode strings. + a generator that yields one item after another as strings. It accepts the same arguments as :meth:`render`. """ - vars = dict(*args, **kwargs) + if self.environment.is_async: + import asyncio + + async def to_list() -> t.List[str]: + return [x async for x in self.generate_async(*args, **kwargs)] + + yield from asyncio.run(to_list()) + return + + ctx = self.new_context(dict(*args, **kwargs)) + try: - for event in self.root_render_func(self.new_context(vars)): - yield event + yield from self.root_render_func(ctx) # type: ignore except Exception: yield self.environment.handle_exception() - def generate_async(self, *args, **kwargs): + async def generate_async( + self, *args: t.Any, **kwargs: t.Any + ) -> t.AsyncIterator[str]: """An async version of :meth:`generate`. Works very similarly but returns an async iterator instead. """ - # see asyncsupport for the actual implementation - raise NotImplementedError( - "This feature is not available for this version of Python" - ) + if not self.environment.is_async: + raise RuntimeError( + "The environment was not created with async mode enabled." + ) + + ctx = self.new_context(dict(*args, **kwargs)) - def new_context(self, vars=None, shared=False, locals=None): + try: + async for event in self.root_render_func(ctx): # type: ignore + yield event + except Exception: + yield self.environment.handle_exception() + + def new_context( + self, + vars: t.Optional[t.Dict[str, t.Any]] = None, + shared: bool = False, + locals: t.Optional[t.Mapping[str, t.Any]] = None, + ) -> Context: """Create a new :class:`Context` for this template. The vars provided will be passed to the template. Per default the globals are added to the context. If shared is set to `True` the data @@ -1145,35 +1389,80 @@ class Template(object): self.environment, self.name, self.blocks, vars, shared, self.globals, locals ) - def make_module(self, vars=None, shared=False, locals=None): + def make_module( + self, + vars: t.Optional[t.Dict[str, t.Any]] = None, + shared: bool = False, + locals: t.Optional[t.Mapping[str, t.Any]] = None, + ) -> "TemplateModule": """This method works like the :attr:`module` attribute when called without arguments but it will evaluate the template on every call rather than caching it. It's also possible to provide a dict which is then used as context. The arguments are the same as for the :meth:`new_context` method. """ - return TemplateModule(self, self.new_context(vars, shared, locals)) + ctx = self.new_context(vars, shared, locals) + return TemplateModule(self, ctx) - def make_module_async(self, vars=None, shared=False, locals=None): + async def make_module_async( + self, + vars: t.Optional[t.Dict[str, t.Any]] = None, + shared: bool = False, + locals: t.Optional[t.Mapping[str, t.Any]] = None, + ) -> "TemplateModule": """As template module creation can invoke template code for asynchronous executions this method must be used instead of the normal :meth:`make_module` one. Likewise the module attribute becomes unavailable in async mode. """ - # see asyncsupport for the actual implementation - raise NotImplementedError( - "This feature is not available for this version of Python" + ctx = self.new_context(vars, shared, locals) + return TemplateModule( + self, ctx, [x async for x in self.root_render_func(ctx)] # type: ignore ) @internalcode - def _get_default_module(self): - if self._module is not None: - return self._module - self._module = rv = self.make_module() - return rv + def _get_default_module(self, ctx: t.Optional[Context] = None) -> "TemplateModule": + """If a context is passed in, this means that the template was + imported. Imported templates have access to the current + template's globals by default, but they can only be accessed via + the context during runtime. + + If there are new globals, we need to create a new module because + the cached module is already rendered and will not have access + to globals from the current context. This new module is not + cached because the template can be imported elsewhere, and it + should have access to only the current template's globals. + """ + if self.environment.is_async: + raise RuntimeError("Module is not available in async mode.") + + if ctx is not None: + keys = ctx.globals_keys - self.globals.keys() + + if keys: + return self.make_module({k: ctx.parent[k] for k in keys}) + + if self._module is None: + self._module = self.make_module() + + return self._module + + async def _get_default_module_async( + self, ctx: t.Optional[Context] = None + ) -> "TemplateModule": + if ctx is not None: + keys = ctx.globals_keys - self.globals.keys() + + if keys: + return await self.make_module_async({k: ctx.parent[k] for k in keys}) + + if self._module is None: + self._module = await self.make_module_async() + + return self._module @property - def module(self): + def module(self) -> "TemplateModule": """The template as module. This is used for imports in the template runtime but is also useful if one wants to access exported template variables from the Python layer: @@ -1188,7 +1477,7 @@ class Template(object): """ return self._get_default_module() - def get_corresponding_lineno(self, lineno): + def get_corresponding_lineno(self, lineno: int) -> int: """Return the source line number of a line number in the generated bytecode as they are not in sync. """ @@ -1198,100 +1487,113 @@ class Template(object): return 1 @property - def is_up_to_date(self): + def is_up_to_date(self) -> bool: """If this variable is `False` there is a newer version available.""" if self._uptodate is None: return True return self._uptodate() @property - def debug_info(self): + def debug_info(self) -> t.List[t.Tuple[int, int]]: """The debug info mapping.""" if self._debug_info: - return [tuple(map(int, x.split("="))) for x in self._debug_info.split("&")] + return [ + tuple(map(int, x.split("="))) # type: ignore + for x in self._debug_info.split("&") + ] + return [] - def __repr__(self): + def __repr__(self) -> str: if self.name is None: - name = "memory:%x" % id(self) + name = f"memory:{id(self):x}" else: name = repr(self.name) - return "<%s %s>" % (self.__class__.__name__, name) + return f"<{type(self).__name__} {name}>" -@implements_to_string -class TemplateModule(object): +class TemplateModule: """Represents an imported template. All the exported names of the template are available as attributes on this object. Additionally - converting it into an unicode- or bytestrings renders the contents. + converting it into a string renders the contents. """ - def __init__(self, template, context, body_stream=None): + def __init__( + self, + template: Template, + context: Context, + body_stream: t.Optional[t.Iterable[str]] = None, + ) -> None: if body_stream is None: if context.environment.is_async: raise RuntimeError( - "Async mode requires a body stream " - "to be passed to a template module. Use " - "the async methods of the API you are " - "using." + "Async mode requires a body stream to be passed to" + " a template module. Use the async methods of the" + " API you are using." ) - body_stream = list(template.root_render_func(context)) + + body_stream = list(template.root_render_func(context)) # type: ignore + self._body_stream = body_stream self.__dict__.update(context.get_exported()) self.__name__ = template.name - def __html__(self): + def __html__(self) -> Markup: return Markup(concat(self._body_stream)) - def __str__(self): + def __str__(self) -> str: return concat(self._body_stream) - def __repr__(self): + def __repr__(self) -> str: if self.__name__ is None: - name = "memory:%x" % id(self) + name = f"memory:{id(self):x}" else: name = repr(self.__name__) - return "<%s %s>" % (self.__class__.__name__, name) + return f"<{type(self).__name__} {name}>" -class TemplateExpression(object): +class TemplateExpression: """The :meth:`jinja2.Environment.compile_expression` method returns an instance of this object. It encapsulates the expression-like access to the template with an expression it wraps. """ - def __init__(self, template, undefined_to_none): + def __init__(self, template: Template, undefined_to_none: bool) -> None: self._template = template self._undefined_to_none = undefined_to_none - def __call__(self, *args, **kwargs): + def __call__(self, *args: t.Any, **kwargs: t.Any) -> t.Optional[t.Any]: context = self._template.new_context(dict(*args, **kwargs)) - consume(self._template.root_render_func(context)) + consume(self._template.root_render_func(context)) # type: ignore rv = context.vars["result"] if self._undefined_to_none and isinstance(rv, Undefined): rv = None return rv -@implements_iterator -class TemplateStream(object): +class TemplateStream: """A template stream works pretty much like an ordinary python generator but it can buffer multiple items to reduce the number of total iterations. Per default the output is unbuffered which means that for every unbuffered - instruction in the template one unicode string is yielded. + instruction in the template one string is yielded. If buffering is enabled with a buffer size of 5, five items are combined - into a new unicode string. This is mainly useful if you are streaming + into a new string. This is mainly useful if you are streaming big templates to a client via WSGI which flushes after each iteration. """ - def __init__(self, gen): + def __init__(self, gen: t.Iterator[str]) -> None: self._gen = gen self.disable_buffering() - def dump(self, fp, encoding=None, errors="strict"): + def dump( + self, + fp: t.Union[str, t.IO], + encoding: t.Optional[str] = None, + errors: t.Optional[str] = "strict", + ) -> None: """Dump the complete stream into a file or file-like object. - Per default unicode strings are written, if you want to encode + Per default strings are written, if you want to encode before writing specify an `encoding`. Example usage:: @@ -1299,16 +1601,19 @@ class TemplateStream(object): Template('Hello {{ name }}!').stream(name='foo').dump('hello.html') """ close = False - if isinstance(fp, string_types): + + if isinstance(fp, str): if encoding is None: encoding = "utf-8" + fp = open(fp, "wb") close = True try: if encoding is not None: - iterable = (x.encode(encoding, errors) for x in self) + iterable = (x.encode(encoding, errors) for x in self) # type: ignore else: - iterable = self + iterable = self # type: ignore + if hasattr(fp, "writelines"): fp.writelines(iterable) else: @@ -1318,17 +1623,17 @@ class TemplateStream(object): if close: fp.close() - def disable_buffering(self): + def disable_buffering(self) -> None: """Disable the output buffering.""" self._next = partial(next, self._gen) self.buffered = False - def _buffered_generator(self, size): - buf = [] + def _buffered_generator(self, size: int) -> t.Iterator[str]: + buf: t.List[str] = [] c_size = 0 push = buf.append - while 1: + while True: try: while c_size < size: c = next(self._gen) @@ -1342,7 +1647,7 @@ class TemplateStream(object): del buf[:] c_size = 0 - def enable_buffering(self, size=5): + def enable_buffering(self, size: int = 5) -> None: """Enable buffering. Buffer `size` items before yielding them.""" if size <= 1: raise ValueError("buffer size too small") @@ -1350,11 +1655,11 @@ class TemplateStream(object): self.buffered = True self._next = partial(next, self._buffered_generator(size)) - def __iter__(self): + def __iter__(self) -> "TemplateStream": return self - def __next__(self): - return self._next() + def __next__(self) -> str: + return self._next() # type: ignore # hook in default template class. if anyone reads this comment: ignore that |