Coverage for lib/datamodel/jinja.py: 99%
79 statements
« prev ^ index » next coverage.py v7.10.1, created at 2025-07-28 07:24 +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, 2024 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, Environment
26from jinja2.environment import Template
27from jinja2.nativetypes import NativeCodeGenerator
28from jinja2.nodes import Output, TemplateData
29from types import GeneratorType
30from typing import Any, Iterable, Optional
33class HermesNotAJinjaExpression(Exception):
34 """Raised when a Jinja statement is found in template"""
37class HermesDataModelAttrsmappingError(Exception):
38 """Raised when an attrsmapping in datamodel is invalid"""
41class HermesTooManyJinjaVarsError(Exception):
42 """Raised when an attrsmapping in datamodel is invalid"""
45class HermesUnknownVarsInJinjaTemplateError(Exception):
46 """Raised when an unknown var is found in a Jinja template"""
49def hermes_native_concat(values: Iterable[Any]) -> Optional[Any]:
50 """Copy of jinja2.nativetypes.native_concat that will return the raw string
51 if the resulting value would have been a complex number
52 """
53 head = list(islice(values, 2))
55 if not head:
56 return None
58 if len(head) == 1:
59 raw = head[0]
60 if not isinstance(raw, str):
61 return raw
62 else:
63 if isinstance(values, GeneratorType):
64 values = chain(head, values)
65 raw = "".join([str(v) for v in values])
67 try:
68 res = literal_eval(
69 # In Python 3.10+ ast.literal_eval removes leading spaces/tabs
70 # from the given string. For backwards compatibility we need to
71 # parse the string ourselves without removing leading spaces/tabs.
72 parse(raw, mode="eval")
73 )
74 except (ValueError, SyntaxError, MemoryError):
75 return raw
77 if isinstance(res, complex):
78 # Return the raw string instead of the evaluated value
79 return raw
80 else:
81 return res
84class HermesNativeEnvironment(Environment):
85 """An environment that renders templates to native Python types, excepted
86 the complex numbers that are ignored."""
88 code_generator_class = NativeCodeGenerator
89 concat = staticmethod(hermes_native_concat) # type: ignore
92class Jinja:
93 """Helper class to compile Jinja expressions, and render query vars"""
95 @classmethod
96 def _compileIfJinjaTemplate(
97 cls,
98 tpl: str,
99 jinjaenv: HermesNativeEnvironment,
100 errorcontext: str,
101 allowOnlyOneTemplate: bool,
102 allowOnlyOneVar: bool,
103 ) -> tuple[Template | str, list[str]]:
104 """Parse specified string to determine if it contains some Jinja or not.
105 Return a tuple (jinjaCompiledTemplate, varlist)
107 If tpl contains some Jinja:
108 - jinjaCompiledTemplate will be a Template instance, to call with
109 .render(contextdict)
110 - varlist will be a list of var names required to render templates
111 else:
112 - jinjaCompiledTemplate will be tpl
113 - varlist will be a list containing only tpl
115 errorcontext: is a string that will prefix error messages
116 allowOnlyOneTemplate: if True, if tpl contains something else than a
117 non-jinja string OR a single template, an HermesDataModelAttrsmappingError
118 will be raised
119 allowOnlyOneVar: if True, if tpl contains more than one variable, an
120 HermesTooManyJinjaVarsError will be raised
121 """
122 env = HermesNativeEnvironment()
123 env.filters.update(jinjaenv.filters)
124 ast = env.parse(tpl)
125 vars = meta.find_undeclared_variables(ast)
127 if len(ast.body) == 0:
128 raise HermesDataModelAttrsmappingError(
129 f"{errorcontext}: Empty value was found"
130 )
132 elif len(ast.body) > 1:
133 if allowOnlyOneTemplate:
134 raise HermesDataModelAttrsmappingError(
135 f"{errorcontext}: Multiple jinja templates found in '''{tpl}''',"
136 " only one is allowed"
137 )
138 else:
139 if not isinstance(ast.body[0], Output):
140 raise HermesNotAJinjaExpression(
141 f"{errorcontext}: Only Jinja expressions '{{{{ ... }}}}' are"
142 f" allowed. Another type of Jinja data was found in '''{tpl}'''"
143 )
145 if len(ast.body[0].nodes) == 1 and isinstance(
146 ast.body[0].nodes[0], TemplateData
147 ):
148 # tpl is not a Jinja template, return it as is
149 return (tpl, [tpl])
151 for item in ast.body[0].nodes:
152 if allowOnlyOneTemplate and isinstance(item, TemplateData):
153 raise HermesDataModelAttrsmappingError(
154 f"{errorcontext}: A mix between jinja templates and raw data"
155 f" was found in '''{tpl}''', with this configuration it's"
156 " impossible to determine source attribute name"
157 )
159 # tpl is a Jinja template, return each var name it contains
160 if allowOnlyOneVar and len(vars) > 1:
161 raise HermesTooManyJinjaVarsError(
162 f"{errorcontext}: {len(vars)} variables found in Jinja template"
163 f" '''{tpl}'''. Only one Jinja var is allowed to ensure data"
164 " consistency"
165 )
167 return (jinjaenv.from_string(tpl), vars)
169 @classmethod
170 def compileIfJinjaTemplate(
171 cls,
172 var: Any,
173 flatvars_set: set[str] | None,
174 jinjaenv: HermesNativeEnvironment,
175 errorcontext: str,
176 allowOnlyOneTemplate: bool,
177 allowOnlyOneVar: bool,
178 excludeFlatVars: set[str] = set(),
179 ) -> Any:
180 """Recursive copy of specified var to replace all jinja templates strings by
181 their compiled template instance.
183 If flatvars_set is specified, every vars met (raw string, or Jinja vars) will be
184 added to it, excepted those specified in excludeFlatVars
186 Returns the same var as specified, where all strings containing jinja templates
187 have been replaced by a compiled version of the template
188 (jinja2.environment.Template instance).
190 errorcontext: is a string that will prefix error messages
191 allowOnlyOneTemplate: if True, if tpl contains something else than a
192 non-jinja string OR a single template, an HermesDataModelAttrsmappingError
193 will be raised
194 allowOnlyOneVar: if True, if tpl contains more than one variable, an
195 HermesTooManyJinjaVarsError will be raised
196 """
197 if type(var) is str:
198 template, varlist = cls._compileIfJinjaTemplate(
199 var, jinjaenv, errorcontext, allowOnlyOneTemplate, allowOnlyOneVar
200 )
201 if type(flatvars_set) is set:
202 flatvars_set.update(set(varlist) - excludeFlatVars)
203 return template
204 elif type(var) is dict:
205 res = {}
206 for k, v in var.items():
207 res[k] = cls.compileIfJinjaTemplate(
208 v,
209 flatvars_set,
210 jinjaenv,
211 errorcontext,
212 allowOnlyOneTemplate,
213 allowOnlyOneVar,
214 excludeFlatVars,
215 )
216 return res
217 elif type(var) is list:
218 return [
219 cls.compileIfJinjaTemplate(
220 v,
221 flatvars_set,
222 jinjaenv,
223 errorcontext,
224 allowOnlyOneTemplate,
225 allowOnlyOneVar,
226 excludeFlatVars,
227 )
228 for v in var
229 ]
230 else:
231 return var
233 @classmethod
234 def renderQueryVars(cls, queryvars: Any, context: dict[str, Any]) -> Any:
235 """Render Jinja queryvars templates with specified context dict, and returns
236 rendered dict"""
237 if isinstance(queryvars, Template):
238 return queryvars.render(context)
239 elif type(queryvars) is dict:
240 return {k: cls.renderQueryVars(v, context) for k, v in queryvars.items()}
241 elif type(queryvars) is list:
242 return [cls.renderQueryVars(v, context) for v in queryvars]
243 else:
244 return queryvars