Coverage for lib/datamodel/jinja.py: 98%
84 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-11 15:35 +0000
1#!/usr/bin/env python3
2# -*- coding: utf-8 -*-
4# Hermes : Change Data Capture (CDC) tool from any source(s) to any target
5# Copyright (C) 2023 INSA Strasbourg
6#
7# This file is part of Hermes.
8#
9# Hermes is free software: you can redistribute it and/or modify
10# it under the terms of the GNU General Public License as published by
11# the Free Software Foundation, either version 3 of the License, or
12# (at your option) any later version.
13#
14# Hermes is distributed in the hope that it will be useful,
15# but WITHOUT ANY WARRANTY; without even the implied warranty of
16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17# GNU General Public License for more details.
18#
19# You should have received a copy of the GNU General Public License
20# along with Hermes. If not, see <https://www.gnu.org/licenses/>.
23from ast import parse, literal_eval
24from itertools import chain, islice
25from jinja2 import meta
26from jinja2.environment import Template
27from jinja2.nativetypes import NativeCodeGenerator, NativeTemplate
28from jinja2.nodes import Output, TemplateData
29from jinja2.sandbox import SandboxedEnvironment
30from types import GeneratorType
31from typing import Any, Iterable, Optional
34class HermesNotAJinjaExpression(Exception):
35 """Raised when a Jinja statement is found in template"""
38class HermesDataModelAttrsmappingError(Exception):
39 """Raised when an attrsmapping in datamodel is invalid"""
42class HermesTooManyJinjaVarsError(Exception):
43 """Raised when an attrsmapping in datamodel is invalid"""
46class HermesUnknownVarsInJinjaTemplateError(Exception):
47 """Raised when an unknown var is found in a Jinja template"""
50def hermes_native_concat(values: Iterable[Any]) -> Optional[Any]:
51 """Copy of jinja2.nativetypes.native_concat that will return the resulting native
52 Python value
53 """
54 head = list(islice(values, 2))
56 if not head:
57 return None
59 if len(head) == 1:
60 return head[0]
61 else:
62 if isinstance(values, GeneratorType):
63 values = chain(head, values)
64 raw = "".join([str(v) for v in values])
66 try:
67 return literal_eval(
68 # In Python 3.10+ ast.literal_eval removes leading spaces/tabs
69 # from the given string. For backwards compatibility we need to
70 # parse the string ourselves without removing leading spaces/tabs.
71 parse(raw, mode="eval")
72 )
73 except (ValueError, SyntaxError, MemoryError):
74 return raw
77class HermesCodeGenerator(NativeCodeGenerator):
78 # Code copied from Ansible - thanks to them !
79 # https://github.com/ansible/ansible/blob/v2.20.4/lib/ansible/_internal/_templating/_jinja_bits.py#L359
80 def _output_const_repr(self, group: Iterable[Any]) -> str:
81 """
82 Prevent Jinja's code generation from stringifying single nodes before
83 generating its repr.
84 This complements the behavioral change in HermesNativeEnvironment.concat which
85 returns single nodes without stringifying them.
86 """
87 # DTFIX-FUTURE: contribute this upstream as a fix to Jinja's native support
88 group_list = list(group)
90 if len(group_list) == 1:
91 return repr(group_list[0])
93 # NB: This is slightly more efficient than Jinja's _output_const_repr, which
94 # generates a throw-away list instance to pass to join.
95 # Before removing this, ensure that upstream Jinja has this change.
96 return repr("".join(map(str, group_list)))
99class HermesNativeEnvironment(SandboxedEnvironment):
100 """An environment that renders templates to native Python types"""
102 code_generator_class = HermesCodeGenerator
103 concat = staticmethod(hermes_native_concat) # type: ignore
106class HermesNativeTemplate(NativeTemplate):
107 environment_class = HermesNativeEnvironment
110HermesNativeEnvironment.template_class = HermesNativeTemplate
113class Jinja:
114 """Helper class to compile Jinja expressions, and render query vars"""
116 @classmethod
117 def _compileIfJinjaTemplate(
118 cls,
119 tpl: str,
120 jinjaenv: HermesNativeEnvironment,
121 errorcontext: str,
122 allowOnlyOneTemplate: bool,
123 allowOnlyOneVar: bool,
124 ) -> tuple[Template | str, list[str]]:
125 """Parse specified string to determine if it contains some Jinja or not.
126 Return a tuple (jinjaCompiledTemplate, varlist)
128 If tpl contains some Jinja:
129 - jinjaCompiledTemplate will be a Template instance, to call with
130 .render(contextdict)
131 - varlist will be a list of var names required to render templates
132 else:
133 - jinjaCompiledTemplate will be tpl
134 - varlist will be a list containing only tpl
136 errorcontext: is a string that will prefix error messages
137 allowOnlyOneTemplate: if True, if tpl contains something else than a
138 non-jinja string OR a single template, an HermesDataModelAttrsmappingError
139 will be raised
140 allowOnlyOneVar: if True, if tpl contains more than one variable, an
141 HermesTooManyJinjaVarsError will be raised
142 """
143 env = HermesNativeEnvironment()
144 env.filters.update(jinjaenv.filters)
145 ast = env.parse(tpl)
146 vars = meta.find_undeclared_variables(ast)
148 if len(ast.body) == 0:
149 raise HermesDataModelAttrsmappingError(
150 f"{errorcontext}: Empty value was found"
151 )
153 elif len(ast.body) > 1:
154 if allowOnlyOneTemplate:
155 raise HermesDataModelAttrsmappingError(
156 f"{errorcontext}: Multiple jinja templates found in '''{tpl}''',"
157 " only one is allowed"
158 )
159 else:
160 if not isinstance(ast.body[0], Output):
161 raise HermesNotAJinjaExpression(
162 f"{errorcontext}: Only Jinja expressions '{{{{ ... }}}}' are"
163 f" allowed. Another type of Jinja data was found in '''{tpl}'''"
164 )
166 if len(ast.body[0].nodes) == 1 and isinstance(
167 ast.body[0].nodes[0], TemplateData
168 ):
169 # tpl is not a Jinja template, return it as is
170 return (tpl, [tpl])
172 for item in ast.body[0].nodes:
173 if allowOnlyOneTemplate and isinstance(item, TemplateData):
174 raise HermesDataModelAttrsmappingError(
175 f"{errorcontext}: A mix between jinja templates and raw data"
176 f" was found in '''{tpl}''', with this configuration it's"
177 " impossible to determine source attribute name"
178 )
180 # tpl is a Jinja template, return each var name it contains
181 if allowOnlyOneVar and len(vars) > 1:
182 raise HermesTooManyJinjaVarsError(
183 f"{errorcontext}: {len(vars)} variables found in Jinja template"
184 f" '''{tpl}'''. Only one Jinja var is allowed to ensure data"
185 " consistency"
186 )
188 return (jinjaenv.from_string(tpl), vars)
190 @classmethod
191 def compileIfJinjaTemplate(
192 cls,
193 var: Any,
194 flatvars_set: set[str] | None,
195 jinjaenv: HermesNativeEnvironment,
196 errorcontext: str,
197 allowOnlyOneTemplate: bool,
198 allowOnlyOneVar: bool,
199 excludeFlatVars: set[str] = set(),
200 ) -> Any:
201 """Recursive copy of specified var to replace all jinja templates strings by
202 their compiled template instance.
204 If flatvars_set is specified, every vars met (raw string, or Jinja vars) will be
205 added to it, excepted those specified in excludeFlatVars
207 Returns the same var as specified, where all strings containing jinja templates
208 have been replaced by a compiled version of the template
209 (jinja2.environment.Template instance).
211 errorcontext: is a string that will prefix error messages
212 allowOnlyOneTemplate: if True, if tpl contains something else than a
213 non-jinja string OR a single template, an HermesDataModelAttrsmappingError
214 will be raised
215 allowOnlyOneVar: if True, if tpl contains more than one variable, an
216 HermesTooManyJinjaVarsError will be raised
217 """
218 if type(var) is str:
219 template, varlist = cls._compileIfJinjaTemplate(
220 var, jinjaenv, errorcontext, allowOnlyOneTemplate, allowOnlyOneVar
221 )
222 if type(flatvars_set) is set:
223 flatvars_set.update(set(varlist) - excludeFlatVars)
224 return template
225 elif type(var) is dict:
226 res = {}
227 for k, v in var.items():
228 res[k] = cls.compileIfJinjaTemplate(
229 v,
230 flatvars_set,
231 jinjaenv,
232 errorcontext,
233 allowOnlyOneTemplate,
234 allowOnlyOneVar,
235 excludeFlatVars,
236 )
237 return res
238 elif type(var) is list:
239 return [
240 cls.compileIfJinjaTemplate(
241 v,
242 flatvars_set,
243 jinjaenv,
244 errorcontext,
245 allowOnlyOneTemplate,
246 allowOnlyOneVar,
247 excludeFlatVars,
248 )
249 for v in var
250 ]
251 else:
252 return var
254 @classmethod
255 def renderQueryVars(cls, queryvars: Any, context: dict[str, Any]) -> Any:
256 """Render Jinja queryvars templates with specified context dict, and returns
257 rendered dict"""
258 if isinstance(queryvars, Template):
259 return queryvars.render(context)
260 elif type(queryvars) is dict:
261 return {k: cls.renderQueryVars(v, context) for k, v in queryvars.items()}
262 elif type(queryvars) is list:
263 return [cls.renderQueryVars(v, context) for v in queryvars]
264 else:
265 return queryvars