414 lines
17 KiB
Python
414 lines
17 KiB
Python
from __future__ import annotations as _annotations
|
|
|
|
import contextlib as _contextlib
|
|
import io as _io
|
|
import os as _os
|
|
import typing as _t
|
|
|
|
from jinja2 import environment as _environment
|
|
|
|
from ansible import _internal
|
|
from ansible import errors as _errors
|
|
from ansible._internal._datatag import _tags, _wrappers
|
|
from ansible._internal._templating import _jinja_bits, _engine, _jinja_common, _template_vars
|
|
|
|
from ansible.module_utils import datatag as _module_utils_datatag
|
|
from ansible.utils.display import Display as _Display
|
|
|
|
if _t.TYPE_CHECKING: # pragma: nocover
|
|
import collections as _collections
|
|
|
|
from ansible.parsing import dataloader as _dataloader
|
|
|
|
_VariableContainer = dict[str, _t.Any] | _collections.ChainMap[str, _t.Any]
|
|
|
|
|
|
_display: _t.Final[_Display] = _Display()
|
|
_UNSET = _t.cast(_t.Any, object())
|
|
_TTrustable = _t.TypeVar('_TTrustable', bound=str | _io.IOBase | _t.TextIO | _t.BinaryIO)
|
|
_TRUSTABLE_TYPES = (str, _io.IOBase)
|
|
|
|
AnsibleUndefined = _jinja_common.UndefinedMarker
|
|
"""Backwards compatibility alias for UndefinedMarker."""
|
|
|
|
|
|
class Templar:
|
|
"""Primary public API container for Ansible's template engine."""
|
|
|
|
def __init__(
|
|
self,
|
|
loader: _dataloader.DataLoader | None = None,
|
|
variables: _VariableContainer | None = None,
|
|
) -> None:
|
|
self._engine = _engine.TemplateEngine(loader=loader, variables=variables)
|
|
self._overrides = _jinja_bits.TemplateOverrides.DEFAULT
|
|
|
|
@classmethod
|
|
@_internal.experimental
|
|
def _from_template_engine(cls, engine: _engine.TemplateEngine) -> _t.Self:
|
|
"""
|
|
EXPERIMENTAL: For internal use within ansible-core only.
|
|
Create a `Templar` instance from the given `TemplateEngine` instance.
|
|
"""
|
|
templar = object.__new__(cls)
|
|
templar._engine = engine.copy()
|
|
templar._overrides = _jinja_bits.TemplateOverrides.DEFAULT
|
|
|
|
return templar
|
|
|
|
def resolve_variable_expression(
|
|
self,
|
|
expression: str,
|
|
*,
|
|
local_variables: dict[str, _t.Any] | None = None,
|
|
) -> _t.Any:
|
|
"""
|
|
Resolve a potentially untrusted string variable expression consisting only of valid identifiers, integers, dots, and indexing containing these.
|
|
Optional local variables may be provided, which can only be referenced directly by the given expression.
|
|
Valid: x, x.y, x[y].z, x[1], 1, x[y.z]
|
|
Error: 'x', x['y'], q('env')
|
|
"""
|
|
return self._engine.resolve_variable_expression(expression, local_variables=local_variables)
|
|
|
|
def evaluate_expression(
|
|
self,
|
|
expression: str,
|
|
*,
|
|
local_variables: dict[str, _t.Any] | None = None,
|
|
escape_backslashes: bool = True,
|
|
) -> _t.Any:
|
|
"""
|
|
Evaluate a trusted string expression and return its result.
|
|
Optional local variables may be provided, which can only be referenced directly by the given expression.
|
|
"""
|
|
return self._engine.evaluate_expression(expression, local_variables=local_variables, escape_backslashes=escape_backslashes)
|
|
|
|
def evaluate_conditional(self, conditional: str | bool) -> bool:
|
|
"""
|
|
Evaluate a trusted string expression or boolean and return its boolean result. A non-boolean result will raise `AnsibleBrokenConditionalError`.
|
|
The ALLOW_BROKEN_CONDITIONALS configuration option can temporarily relax this requirement, allowing truthy conditionals to succeed.
|
|
The ALLOW_EMBEDDED_TEMPLATES configuration option can temporarily enable inline Jinja template delimiter support (e.g., {{ }}, {% %}).
|
|
"""
|
|
return self._engine.evaluate_conditional(conditional)
|
|
|
|
@property
|
|
def basedir(self) -> str:
|
|
"""The basedir from DataLoader."""
|
|
# DTFIX-FUTURE: come up with a better way to handle this so it can be deprecated
|
|
return self._engine.basedir
|
|
|
|
@property
|
|
def available_variables(self) -> _VariableContainer:
|
|
"""Available variables this instance will use when templating."""
|
|
return self._engine.available_variables
|
|
|
|
@available_variables.setter
|
|
def available_variables(self, variables: _VariableContainer) -> None:
|
|
self._engine.available_variables = variables
|
|
|
|
@property
|
|
def _available_variables(self) -> _VariableContainer:
|
|
"""Deprecated. Use `available_variables` instead."""
|
|
# Commonly abused by numerous collection lookup plugins and the Ceph Ansible `config_template` action.
|
|
_display.deprecated(
|
|
msg='Direct access to the `_available_variables` internal attribute is deprecated.',
|
|
help_text='Use `available_variables` instead.',
|
|
version='2.23',
|
|
)
|
|
|
|
return self.available_variables
|
|
|
|
@property
|
|
def _loader(self) -> _dataloader.DataLoader:
|
|
"""Deprecated. Use `copy_with_new_env` to create a new instance."""
|
|
# Abused by cloud.common, community.general and felixfontein.tools collections to create a new Templar instance.
|
|
_display.deprecated(
|
|
msg='Direct access to the `_loader` internal attribute is deprecated.',
|
|
help_text='Use `copy_with_new_env` to create a new instance.',
|
|
version='2.23',
|
|
)
|
|
|
|
return self._engine._loader
|
|
|
|
@property
|
|
def environment(self) -> _environment.Environment:
|
|
"""Deprecated."""
|
|
_display.deprecated(
|
|
msg='Direct access to the `environment` attribute is deprecated.',
|
|
help_text='Consider using `copy_with_new_env` or passing `overrides` to `template`.',
|
|
version='2.23',
|
|
)
|
|
|
|
return self._engine.environment
|
|
|
|
def copy_with_new_env(
|
|
self,
|
|
*,
|
|
searchpath: str | _os.PathLike | _t.Sequence[str | _os.PathLike] | None = None,
|
|
available_variables: _VariableContainer | None = None,
|
|
**context_overrides: _t.Any,
|
|
) -> Templar:
|
|
"""Return a new templar based on the current one with customizations applied."""
|
|
if context_overrides.pop('environment_class', _UNSET) is not _UNSET:
|
|
_display.deprecated(
|
|
msg="The `environment_class` argument is ignored.",
|
|
version='2.23',
|
|
)
|
|
|
|
if context_overrides:
|
|
_display.deprecated(
|
|
msg='Passing Jinja environment overrides to `copy_with_new_env` is deprecated.',
|
|
help_text='Pass Jinja environment overrides to individual `template` calls.',
|
|
version='2.23',
|
|
)
|
|
|
|
templar = Templar(
|
|
loader=self._engine._loader,
|
|
variables=self._engine._variables if available_variables is None else available_variables,
|
|
)
|
|
|
|
# backward compatibility: filter out None values from overrides, even though it is a valid value for some of them
|
|
templar._overrides = self._overrides.merge({key: value for key, value in context_overrides.items() if value is not None})
|
|
|
|
if searchpath is not None:
|
|
templar._engine.environment.loader.searchpath = searchpath
|
|
|
|
return templar
|
|
|
|
@_contextlib.contextmanager
|
|
def set_temporary_context(
|
|
self,
|
|
*,
|
|
searchpath: str | _os.PathLike | _t.Sequence[str | _os.PathLike] | None = None,
|
|
available_variables: _VariableContainer | None = None,
|
|
**context_overrides: _t.Any,
|
|
) -> _t.Generator[None, None, None]:
|
|
"""Context manager used to set temporary templating context, without having to worry about resetting original values afterward."""
|
|
_display.deprecated(
|
|
msg='The `set_temporary_context` method on `Templar` is deprecated.',
|
|
help_text='Use the `copy_with_new_env` method on `Templar` instead.',
|
|
version='2.23',
|
|
)
|
|
|
|
targets = dict(
|
|
searchpath=self._engine.environment.loader,
|
|
available_variables=self._engine,
|
|
)
|
|
|
|
target_args = dict(
|
|
searchpath=searchpath,
|
|
available_variables=available_variables,
|
|
)
|
|
|
|
original: dict[str, _t.Any] = {}
|
|
previous_overrides = self._overrides
|
|
|
|
try:
|
|
for key, value in target_args.items():
|
|
if value is not None:
|
|
target = targets[key]
|
|
original[key] = getattr(target, key)
|
|
setattr(target, key, value)
|
|
|
|
# backward compatibility: filter out None values from overrides, even though it is a valid value for some of them
|
|
self._overrides = self._overrides.merge({key: value for key, value in context_overrides.items() if value is not None})
|
|
|
|
yield
|
|
finally:
|
|
for key, value in original.items():
|
|
setattr(targets[key], key, value)
|
|
|
|
self._overrides = previous_overrides
|
|
|
|
# noinspection PyUnusedLocal
|
|
def template(
|
|
self,
|
|
variable: _t.Any,
|
|
convert_bare: bool = _UNSET,
|
|
preserve_trailing_newlines: bool = True,
|
|
escape_backslashes: bool = True,
|
|
fail_on_undefined: bool = True,
|
|
overrides: dict[str, _t.Any] | None = None,
|
|
convert_data: bool = _UNSET,
|
|
disable_lookups: bool = _UNSET,
|
|
) -> _t.Any:
|
|
"""Templates (possibly recursively) any given data as input."""
|
|
# DTFIX-FUTURE: offer a public version of TemplateOverrides to support an optional strongly typed `overrides` argument
|
|
if convert_bare is not _UNSET:
|
|
# Skipping a deferred deprecation due to minimal usage outside ansible-core.
|
|
# Use `hasattr(templar, 'evaluate_expression')` to determine if `template` or `evaluate_expression` should be used.
|
|
_display.deprecated(
|
|
msg="Passing `convert_bare` to `template` is deprecated.",
|
|
help_text="Use `evaluate_expression` instead.",
|
|
version="2.23",
|
|
)
|
|
|
|
if convert_bare and isinstance(variable, str):
|
|
contains_filters = "|" in variable
|
|
first_part = variable.split("|")[0].split(".")[0].split("[")[0]
|
|
convert_bare = (contains_filters or first_part in self.available_variables) and not self.is_possibly_template(variable, overrides)
|
|
else:
|
|
convert_bare = False
|
|
else:
|
|
convert_bare = False
|
|
|
|
if fail_on_undefined is None:
|
|
# The pre-2.19 config fallback is ignored for content portability.
|
|
_display.deprecated(
|
|
msg="Falling back to `True` for `fail_on_undefined`.",
|
|
help_text="Use either `True` or `False` for `fail_on_undefined` when calling `template`.",
|
|
version="2.23",
|
|
)
|
|
|
|
fail_on_undefined = True
|
|
|
|
if convert_data is not _UNSET:
|
|
# Skipping a deferred deprecation due to minimal usage outside ansible-core.
|
|
# Use `hasattr(templar, 'evaluate_expression')` as a surrogate check to determine if `convert_data` is accepted.
|
|
_display.deprecated(
|
|
msg="Passing `convert_data` to `template` is deprecated.",
|
|
version="2.23",
|
|
)
|
|
|
|
if disable_lookups is not _UNSET:
|
|
# Skipping a deferred deprecation due to no known usage outside ansible-core.
|
|
# Use `hasattr(templar, 'evaluate_expression')` as a surrogate check to determine if `disable_lookups` is accepted.
|
|
_display.deprecated(
|
|
msg="Passing `disable_lookups` to `template` is deprecated.",
|
|
version="2.23",
|
|
)
|
|
|
|
try:
|
|
if convert_bare: # pre-2.19 compat
|
|
return self.evaluate_expression(variable, escape_backslashes=escape_backslashes)
|
|
|
|
return self._engine.template(
|
|
variable=variable,
|
|
options=_engine.TemplateOptions(
|
|
preserve_trailing_newlines=preserve_trailing_newlines,
|
|
escape_backslashes=escape_backslashes,
|
|
overrides=self._overrides.merge(overrides),
|
|
),
|
|
mode=_engine.TemplateMode.ALWAYS_FINALIZE,
|
|
)
|
|
except _errors.AnsibleUndefinedVariable:
|
|
if not fail_on_undefined:
|
|
return variable
|
|
|
|
raise
|
|
|
|
def is_template(self, data: _t.Any) -> bool:
|
|
"""
|
|
Evaluate the input data to determine if it contains a template, even if that template is invalid. Containers will be recursively searched.
|
|
Objects subject to template-time transforms that do not yield a template are not considered templates by this method.
|
|
Gating a conditional call to `template` with this method is redundant and inefficient -- request templating unconditionally instead.
|
|
"""
|
|
return self._engine.is_template(data, self._overrides)
|
|
|
|
def is_possibly_template(
|
|
self,
|
|
data: _t.Any,
|
|
overrides: dict[str, _t.Any] | None = None,
|
|
) -> bool:
|
|
"""
|
|
A lightweight check to determine if the given value is a string that looks like it contains a template, even if that template is invalid.
|
|
Returns `True` if the given value is a string that starts with a Jinja overrides header or if it contains template start strings.
|
|
Gating a conditional call to `template` with this method is redundant and inefficient -- request templating unconditionally instead.
|
|
"""
|
|
return isinstance(data, str) and _jinja_bits.is_possibly_template(data, self._overrides.merge(overrides))
|
|
|
|
def do_template(
|
|
self,
|
|
data: _t.Any,
|
|
preserve_trailing_newlines: bool = True,
|
|
escape_backslashes: bool = True,
|
|
fail_on_undefined: bool = True,
|
|
overrides: dict[str, _t.Any] | None = None,
|
|
disable_lookups: bool = _UNSET,
|
|
convert_data: bool = _UNSET,
|
|
) -> _t.Any:
|
|
"""Deprecated. Use `template` instead."""
|
|
_display.deprecated(
|
|
msg='The `do_template` method on `Templar` is deprecated.',
|
|
help_text='Use the `template` method on `Templar` instead.',
|
|
version='2.23',
|
|
)
|
|
|
|
if not isinstance(data, str):
|
|
return data
|
|
|
|
return self.template(
|
|
variable=data,
|
|
preserve_trailing_newlines=preserve_trailing_newlines,
|
|
escape_backslashes=escape_backslashes,
|
|
fail_on_undefined=fail_on_undefined,
|
|
overrides=overrides,
|
|
disable_lookups=disable_lookups,
|
|
convert_data=convert_data,
|
|
)
|
|
|
|
|
|
def generate_ansible_template_vars(
|
|
path: str,
|
|
fullpath: str | None = None,
|
|
dest_path: str | None = None,
|
|
) -> dict[str, object]:
|
|
"""
|
|
Generate and return a dictionary with variable metadata about the template specified by `fullpath`.
|
|
If `fullpath` is `None`, `path` will be used instead.
|
|
"""
|
|
# deprecated description="deprecate `generate_ansible_template_vars`, collections should inline the necessary variables" core_version="2.23"
|
|
return _template_vars.generate_ansible_template_vars(path=path, fullpath=fullpath, dest_path=dest_path, include_ansible_managed=True)
|
|
|
|
|
|
def trust_as_template(value: _TTrustable) -> _TTrustable:
|
|
"""
|
|
Returns `value` tagged as trusted for templating.
|
|
Raises a `TypeError` if `value` is not a supported type.
|
|
"""
|
|
if isinstance(value, str):
|
|
return _tags.TrustedAsTemplate().tag(value) # type: ignore[return-value]
|
|
|
|
if isinstance(value, _io.IOBase): # covers TextIO and BinaryIO at runtime, but type checking disagrees
|
|
return _wrappers.TaggedStreamWrapper(value, _tags.TrustedAsTemplate())
|
|
|
|
raise TypeError(f"Trust cannot be applied to {_module_utils_datatag.native_type_name(value)}, only to 'str' or 'IOBase'.")
|
|
|
|
|
|
def is_trusted_as_template(value: object) -> bool:
|
|
"""
|
|
Returns `True` if `value` is a `str` or `IOBase` marked as trusted for templating, otherwise returns `False`.
|
|
Returns `False` for types which cannot be trusted for templating.
|
|
Containers are not recursed and will always return `False`.
|
|
This function should not be needed for production code, but may be useful in unit tests.
|
|
"""
|
|
return isinstance(value, _TRUSTABLE_TYPES) and _tags.TrustedAsTemplate.is_tagged_on(value)
|
|
|
|
|
|
_TCallable = _t.TypeVar('_TCallable', bound=_t.Callable)
|
|
|
|
|
|
def accept_args_markers(plugin: _TCallable) -> _TCallable:
|
|
"""
|
|
A decorator to mark a Jinja plugin as capable of handling `Marker` values for its top-level arguments.
|
|
Non-decorated plugin invocation is skipped when a top-level argument is a `Marker`, with the first such value substituted as the plugin result.
|
|
This ensures that only plugins which understand `Marker` instances for top-level arguments will encounter them.
|
|
"""
|
|
plugin.accept_args_markers = True
|
|
|
|
return plugin
|
|
|
|
|
|
def accept_lazy_markers(plugin: _TCallable) -> _TCallable:
|
|
"""
|
|
A decorator to mark a Jinja plugin as capable of handling `Marker` values retrieved from lazy containers.
|
|
Non-decorated plugins will trigger a `MarkerError` exception when attempting to retrieve a `Marker` from a lazy container.
|
|
This ensures that only plugins which understand lazy retrieval of `Marker` instances will encounter them.
|
|
"""
|
|
plugin.accept_lazy_markers = True
|
|
|
|
return plugin
|
|
|
|
|
|
get_first_marker_arg = _jinja_common.get_first_marker_arg
|