sqlmesh.utils.jinja

  1from __future__ import annotations
  2
  3import base64
  4import importlib
  5import json
  6import re
  7import typing as t
  8import zlib
  9from collections import defaultdict
 10from enum import Enum
 11from sys import exc_info
 12from traceback import walk_tb
 13
 14from jinja2 import Environment, Template, nodes, UndefinedError
 15from jinja2.runtime import Macro
 16from sqlglot import Dialect, Parser, TokenType
 17from sqlglot.expressions import Expression
 18
 19from sqlmesh.core import constants as c
 20from sqlmesh.core import dialect as d
 21from sqlmesh.utils import AttributeDict
 22from sqlmesh.utils.pydantic import PRIVATE_FIELDS, PydanticModel, field_serializer, field_validator
 23from sqlmesh.utils.metaprogramming import SqlValue
 24
 25
 26if t.TYPE_CHECKING:
 27    CallNames = t.Tuple[t.Tuple[str, ...], t.Union[nodes.Call, nodes.Getattr]]
 28
 29SQLMESH_JINJA_PACKAGE = "sqlmesh.utils.jinja"
 30
 31
 32def b64decode(value: t.Union[str, bytes]) -> str:
 33    """Decode a base64-encoded value and return it as UTF-8 text.
 34
 35    Intended for base64-encoded string/JSON secrets (for example a service-account
 36    key stored in an environment variable), not arbitrary binary payloads.
 37    """
 38    decoded = value.encode("utf-8") if isinstance(value, str) else value
 39    return base64.b64decode(decoded).decode("utf-8")
 40
 41
 42def b64encode(value: t.Union[str, bytes]) -> str:
 43    """Base64-encode a value and return the encoding as UTF-8 text.
 44
 45    The input is treated as UTF-8 text, mirroring ``b64decode``; it is intended for
 46    string/JSON secrets rather than arbitrary binary payloads.
 47    """
 48    encoded = value.encode("utf-8") if isinstance(value, str) else value
 49    return base64.b64encode(encoded).decode("utf-8")
 50
 51
 52def create_builtin_filters() -> t.Dict[str, t.Callable]:
 53    return {
 54        "b64decode": b64decode,
 55        "b64encode": b64encode,
 56    }
 57
 58
 59def environment(**kwargs: t.Any) -> Environment:
 60    extensions = kwargs.pop("extensions", [])
 61    extensions.append("jinja2.ext.do")
 62    extensions.append("jinja2.ext.loopcontrols")
 63    env = Environment(extensions=extensions, **kwargs)
 64    env.filters.update(create_builtin_filters())
 65    return env
 66
 67
 68ENVIRONMENT = environment()
 69
 70
 71class MacroReference(PydanticModel, frozen=True):
 72    package: t.Optional[str] = None
 73    name: str
 74
 75    @property
 76    def reference(self) -> str:
 77        if self.package is None:
 78            return self.name
 79        return ".".join((self.package, self.name))
 80
 81    def __str__(self) -> str:
 82        return self.reference
 83
 84
 85class MacroInfo(PydanticModel):
 86    """Class to hold macro and its calls"""
 87
 88    definition: str
 89    depends_on: t.List[MacroReference]
 90    is_top_level: bool = False
 91
 92
 93class MacroReturnVal(Exception):
 94    def __init__(self, val: t.Any):
 95        self.value = val
 96
 97
 98class MacroExtractor(Parser):
 99    def extract(self, jinja: str, dialect: str = "") -> t.Dict[str, MacroInfo]:
100        """Extract a dictionary of macro definitions from a jinja string.
101
102        Args:
103            jinja: The jinja string to extract from.
104            dialect: The dialect of SQL.
105
106        Returns:
107            A dictionary of macro name to macro definition.
108        """
109        self.reset()
110        self.sql = jinja
111        self._tokens = Dialect.get_or_raise(dialect).tokenize(jinja)
112
113        # guard for older sqlglot versions (before 30.0.3)
114        if hasattr(self, "_tokens_size"):
115            # keep the cached length in sync
116            self._tokens_size = len(self._tokens)
117        self._index = -1
118        self._advance()
119
120        macros: t.Dict[str, MacroInfo] = {}
121
122        while self._curr:
123            if self._curr.token_type == TokenType.BLOCK_START:
124                macro_start = self._curr
125            elif self._tag == "MACRO" and self._next:
126                name = self._next.text
127                while self._curr and self._curr.token_type != TokenType.BLOCK_END:
128                    self._advance()
129
130                while self._curr and self._tag != "ENDMACRO":
131                    self._advance()
132
133                macro_str = self._find_sql(macro_start, self._next)
134                macros[name] = MacroInfo(
135                    definition=macro_str,
136                    depends_on=list(extract_macro_references_and_variables(macro_str)[0]),
137                )
138
139            self._advance()
140
141        return macros
142
143    def _advance(self, times: int = 1) -> None:
144        super()._advance(times)
145        self._tag = (
146            self._curr.text.upper()
147            if self._curr and self._prev and self._prev.token_type == TokenType.BLOCK_START
148            else ""
149        )
150
151
152def call_name(node: nodes.Expr) -> t.Tuple[str, ...]:
153    if isinstance(node, nodes.Name):
154        return (node.name,)
155    if isinstance(node, nodes.Const):
156        return (f"'{node.value}'",)
157    if isinstance(node, nodes.Getattr):
158        return call_name(node.node) + (node.attr,)
159    if isinstance(node, (nodes.Getitem, nodes.Call)):
160        return call_name(node.node)
161    return ()
162
163
164def render_jinja(query: str, methods: t.Optional[t.Dict[str, t.Any]] = None) -> str:
165    return ENVIRONMENT.from_string(query).render(methods or {})
166
167
168def find_call_names(node: nodes.Node, vars_in_scope: t.Set[str]) -> t.Iterator[CallNames]:
169    vars_in_scope = vars_in_scope.copy()
170    for child_node in node.iter_child_nodes():
171        if "target" in child_node.fields:
172            # For nodes with assignment targets (Assign, AssignBlock, For, Import),
173            # the target name could shadow a reference in the right hand side.
174            # So we need to process the RHS before adding the target to scope.
175            # For example: {% set model = model.path %} should track model.path.
176            yield from find_call_names(child_node, vars_in_scope)
177
178            target = getattr(child_node, "target")
179            if isinstance(target, nodes.Name):
180                vars_in_scope.add(target.name)
181            elif isinstance(target, nodes.Tuple):
182                for item in target.items:
183                    if isinstance(item, nodes.Name):
184                        vars_in_scope.add(item.name)
185        elif isinstance(child_node, nodes.Macro):
186            for arg in child_node.args:
187                vars_in_scope.add(arg.name)
188        elif isinstance(child_node, nodes.Call) or (
189            isinstance(child_node, nodes.Getattr) and not isinstance(child_node.node, nodes.Getattr)
190        ):
191            name = call_name(child_node)
192            if name[0][0] != "'" and name[0] not in vars_in_scope:
193                yield (name, child_node)
194
195        if "target" not in child_node.fields:
196            yield from find_call_names(child_node, vars_in_scope)
197
198
199def extract_call_names(
200    jinja_str: str, cache: t.Optional[t.Dict[str, t.Tuple[t.List[CallNames], bool]]] = None
201) -> t.List[CallNames]:
202    def parse() -> t.List[CallNames]:
203        return list(find_call_names(ENVIRONMENT.parse(jinja_str), set()))
204
205    if cache is not None:
206        key = str(zlib.crc32(jinja_str.encode("utf-8")))
207        if key in cache:
208            names = cache[key][0]
209        else:
210            names = parse()
211        cache[key] = (names, True)
212        return names
213    return parse()
214
215
216def is_variable_node(n: nodes.Node) -> bool:
217    return (
218        isinstance(n, nodes.Call)
219        and isinstance(n.node, nodes.Name)
220        and n.node.name in (c.VAR, c.BLUEPRINT_VAR)
221    )
222
223
224def extract_macro_references_and_variables(
225    *jinja_strs: str,
226) -> t.Tuple[t.Set[MacroReference], t.Set[str]]:
227    macro_references = set()
228    variables = set()
229    for jinja_str in jinja_strs:
230        for call_name, node in extract_call_names(jinja_str):
231            if call_name[0] in (c.VAR, c.BLUEPRINT_VAR):
232                if not is_variable_node(node):
233                    # Find the variable node which could be nested
234                    for n in node.find_all(nodes.Call):
235                        if is_variable_node(n):
236                            node = n
237                            break
238                    else:
239                        raise ValueError(f"Could not find variable name in {jinja_str}")
240                node = t.cast(nodes.Call, node)
241                args = [jinja_call_arg_name(arg) for arg in node.args]
242                if args and args[0]:
243                    variables.add(args[0].lower())
244            elif call_name[0] == c.GATEWAY:
245                variables.add(c.GATEWAY)
246            elif len(call_name) == 1:
247                macro_references.add(MacroReference(name=call_name[0]))
248            elif len(call_name) == 2:
249                macro_references.add(MacroReference(package=call_name[0], name=call_name[1]))
250    return macro_references, variables
251
252
253def sort_dict_recursive(
254    item: t.Dict[str, t.Any],
255) -> t.Dict[str, t.Any]:
256    sorted_dict: t.Dict[str, t.Any] = {}
257    for k, v in sorted(item.items()):
258        if isinstance(v, list):
259            sorted_dict[k] = sorted(v)
260        elif isinstance(v, dict):
261            sorted_dict[k] = sort_dict_recursive(v)
262        else:
263            sorted_dict[k] = v
264    return sorted_dict
265
266
267JinjaGlobalAttribute = t.Union[str, int, float, bool, AttributeDict]
268
269
270class JinjaMacroRegistry(PydanticModel):
271    """Registry for Jinja macros.
272
273    Args:
274        packages: The mapping from package name to a collection of macro definitions.
275        root_macros: The collection of top-level macro definitions.
276        global_objs: The global objects.
277        create_builtins_module: The name of a module which defines the `create_builtins` factory
278            function that will be used to construct builtin variables and functions.
279        root_package_name: The name of the root package. If specified root macros will be available
280            as both `root_package_name.macro_name` and `macro_name`.
281        top_level_packages: The list of top-level packages. Macros in this packages will be available
282            as both `package_name.macro_name` and `macro_name`.
283    """
284
285    packages: t.Dict[str, t.Dict[str, MacroInfo]] = {}
286    root_macros: t.Dict[str, MacroInfo] = {}
287    global_objs: t.Dict[str, JinjaGlobalAttribute] = {}
288    create_builtins_module: t.Optional[str] = SQLMESH_JINJA_PACKAGE
289    root_package_name: t.Optional[str] = None
290    top_level_packages: t.List[str] = []
291
292    _parser_cache: t.Dict[t.Tuple[t.Optional[str], str], Template] = {}
293    _trimmed: bool = False
294    __environment: t.Optional[Environment] = None
295
296    def __getstate__(self) -> t.Dict[t.Any, t.Any]:
297        state = super().__getstate__()
298        private = state[PRIVATE_FIELDS]
299        private["_parser_cache"] = {}
300        private["_JinjaMacroRegistry__environment"] = None
301        return state
302
303    @field_validator("global_objs", mode="before")
304    @classmethod
305    def _validate_global_objs(cls, value: t.Any) -> t.Any:
306        def _normalize(val: t.Any) -> t.Any:
307            if isinstance(val, dict):
308                return AttributeDict({k: _normalize(v) for k, v in val.items()})
309            if isinstance(val, list):
310                return [_normalize(v) for v in val]
311            if isinstance(val, set):
312                return [_normalize(v) for v in sorted(val)]
313            if isinstance(val, Enum):
314                return val.value
315            return val
316
317        return _normalize(value)
318
319    @field_serializer("global_objs")
320    def _serialize_attribute_dict(
321        self, value: t.Dict[str, JinjaGlobalAttribute]
322    ) -> t.Dict[str, t.Any]:
323        # NOTE: This is called only when used with Pydantic V2.
324        def _convert(
325            val: t.Union[t.Dict[str, JinjaGlobalAttribute], t.Dict[str, t.Any]],
326        ) -> t.Dict[str, t.Any]:
327            return {k: _convert(v) if isinstance(v, AttributeDict) else v for k, v in val.items()}
328
329        return _convert(value)
330
331    @property
332    def trimmed(self) -> bool:
333        return self._trimmed
334
335    def add_macros(self, macros: t.Dict[str, MacroInfo], package: t.Optional[str] = None) -> None:
336        """Adds macros to the target package.
337
338        Args:
339            macros: Macros that should be added.
340            package: The name of the package the given macros belong to. If not specified, the provided
341            macros will be added to the root namespace.
342        """
343
344        if package is not None:
345            package_macros = self.packages.get(package, {})
346            package_macros.update(macros)
347            self.packages[package] = package_macros
348        else:
349            self.root_macros.update(macros)
350
351    def add_globals(self, globals: t.Dict[str, JinjaGlobalAttribute]) -> None:
352        """Adds global objects to the registry.
353
354        Args:
355            globals: The global objects that should be added.
356        """
357        # Keep the registry lightweight when the graph is not needed
358        if not "graph" in self.packages:
359            globals.pop("flat_graph", None)
360        self.global_objs.update(**self._validate_global_objs(globals))
361
362    def build_macro(self, reference: MacroReference, **kwargs: t.Any) -> t.Optional[t.Callable]:
363        """Builds a Python callable for a macro with the given reference.
364
365        Args:
366            reference: The macro reference.
367        Returns:
368            The macro as a Python callable or None if not found.
369        """
370        env: Environment = self.build_environment(**kwargs)
371        if reference.package is not None:
372            package = env.globals.get(reference.package, {})
373            return package.get(reference.name)  # type: ignore
374        return env.globals.get(reference.name)  # type: ignore
375
376    def build_environment(self, **kwargs: t.Any) -> Environment:
377        """Builds a new Jinja environment based on this registry."""
378
379        context: t.Dict[str, t.Any] = {}
380
381        root_macros = {
382            name: self._MacroWrapper(name, None, self, context)
383            for name, macro in self.root_macros.items()
384        }
385
386        package_macros: t.Dict[str, t.Any] = defaultdict(AttributeDict)
387        for package_name, macros in self.packages.items():
388            for macro_name, macro in macros.items():
389                macro_wrapper = self._MacroWrapper(macro_name, package_name, self, context)
390                package_macros[package_name][macro_name] = macro_wrapper
391                if macro.is_top_level and macro_name not in root_macros:
392                    root_macros[macro_name] = macro_wrapper
393
394        top_level_packages = self.top_level_packages.copy()
395
396        if self.root_package_name is not None:
397            package_macros[self.root_package_name].update(root_macros)
398            top_level_packages.append(self.root_package_name)
399
400        env = environment()
401
402        builtin_globals = self._create_builtin_globals(kwargs)
403        for top_level_package_name in top_level_packages:
404            # Make sure that the top-level package doesn't fully override the same builtin package.
405            package_macros[top_level_package_name] = AttributeDict(
406                {
407                    **(builtin_globals.pop(top_level_package_name, None) or {}),
408                    **(package_macros.get(top_level_package_name) or {}),
409                }
410            )
411            root_macros.update(package_macros[top_level_package_name])
412
413        context.update(builtin_globals)
414        context.update(root_macros)
415        context.update(package_macros)
416        context["render"] = lambda input: env.from_string(input).render()
417
418        env.globals.update(context)
419        env.filters.update(self._environment.filters)
420        return env
421
422    def trim(
423        self, dependencies: t.Iterable[MacroReference], package: t.Optional[str] = None
424    ) -> JinjaMacroRegistry:
425        """Trims the registry by keeping only macros with given references and their transitive dependencies.
426
427        Args:
428            dependencies: References to macros that should be kept.
429            package: The name of the package in the context of which the trimming should be performed.
430
431        Returns:
432            A new trimmed registry.
433        """
434        dependencies_by_package: t.Dict[t.Optional[str], t.Set[str]] = defaultdict(set)
435        for dep in dependencies:
436            dependencies_by_package[dep.package or package].add(dep.name)
437
438        top_level_packages = self.top_level_packages.copy()
439        if package is not None:
440            top_level_packages.append(package)
441
442        result = JinjaMacroRegistry(
443            global_objs=self.global_objs.copy(),
444            create_builtins_module=self.create_builtins_module,
445            root_package_name=self.root_package_name,
446            top_level_packages=top_level_packages,
447        )
448        for package, names in dependencies_by_package.items():
449            result = result.merge(self._trim_macros(names, package))
450
451        result._trimmed = True
452
453        return result
454
455    def merge(self, other: JinjaMacroRegistry) -> JinjaMacroRegistry:
456        """Returns a copy of the registry which contains macros from both this and `other` instances.
457
458        Args:
459            other: The other registry instance.
460
461        Returns:
462            A new merged registry.
463        """
464
465        root_macros = {
466            **self.root_macros,
467            **other.root_macros,
468        }
469
470        packages = {}
471        for package in {*self.packages, *other.packages}:
472            packages[package] = {
473                **self.packages.get(package, {}),
474                **other.packages.get(package, {}),
475            }
476
477        global_objs = {
478            **self.global_objs,
479            **other.global_objs,
480        }
481
482        return JinjaMacroRegistry(
483            packages=packages,
484            root_macros=root_macros,
485            global_objs=global_objs,
486            create_builtins_module=self.create_builtins_module or other.create_builtins_module,
487            root_package_name=self.root_package_name or other.root_package_name,
488            top_level_packages=[*self.top_level_packages, *other.top_level_packages],
489        )
490
491    def to_expressions(self) -> t.List[Expression]:
492        output: t.List[Expression] = []
493
494        filtered_objs = {
495            k: v for k, v in self.global_objs.items() if k in ("refs", "sources", "vars")
496        }
497        if filtered_objs:
498            output.append(
499                d.PythonCode(
500                    expressions=[
501                        f"{k} = '{v}'" if isinstance(v, str) else f"{k} = {v}"
502                        for k, v in sort_dict_recursive(filtered_objs).items()
503                    ]
504                )
505            )
506
507        for macro_name, macro_info in sorted(self.root_macros.items()):
508            output.append(d.jinja_statement(macro_info.definition))
509
510        for _, package in sorted(self.packages.items()):
511            for macro_name, macro_info in sorted(package.items()):
512                output.append(d.jinja_statement(macro_info.definition))
513
514        return output
515
516    @property
517    def data_hash_values(self) -> t.List[str]:
518        data = []
519
520        for macro_name, macro in sorted(self.root_macros.items()):
521            data.append(macro_name)
522            data.append(macro.definition)
523
524        for _, package in sorted(self.packages.items()):
525            for macro_name, macro in sorted(package.items()):
526                data.append(macro_name)
527                data.append(macro.definition)
528
529        trimmed_global_objs = {
530            k: self.global_objs[k] for k in ("refs", "sources", "vars") if k in self.global_objs
531        }
532        data.append(json.dumps(trimmed_global_objs, sort_keys=True))
533
534        return data
535
536    def __deepcopy__(self, memo: t.Optional[t.Dict[int, t.Any]] = None) -> JinjaMacroRegistry:
537        return JinjaMacroRegistry.parse_obj(self.dict())
538
539    def _parse_macro(self, name: str, package: t.Optional[str]) -> Template:
540        cache_key = (package, name)
541        if cache_key not in self._parser_cache:
542            macro = self._get_macro(name, package)
543
544            definition: nodes.Template = self._environment.parse(macro.definition)
545            if _is_private_macro(name):
546                # A workaround to expose private jinja macros.
547                definition = self._to_non_private_macro_def(name, definition)
548
549            self._parser_cache[cache_key] = self._environment.from_string(definition)
550        return self._parser_cache[cache_key]
551
552    @property
553    def _environment(self) -> Environment:
554        if self.__environment is None:
555            self.__environment = environment()
556            self.__environment.filters.update(self._create_builtin_filters())
557        return self.__environment
558
559    def _trim_macros(
560        self,
561        names: t.Set[str],
562        package: t.Optional[str] = None,
563        visited: t.Optional[t.Dict[t.Optional[str], t.Set[str]]] = None,
564    ) -> JinjaMacroRegistry:
565        if visited is None:
566            visited = defaultdict(set)
567
568        macros = self.packages.get(package, {}) if package is not None else self.root_macros
569        trimmed_macros = {}
570
571        dependencies: t.Dict[t.Optional[str], t.Set[str]] = defaultdict(set)
572
573        for name in names:
574            if name in macros and name not in visited[package]:
575                macro = macros[name]
576                trimmed_macros[name] = macro
577                for dependency in macro.depends_on:
578                    dependencies[dependency.package or package].add(dependency.name)
579                visited[package].add(name)
580
581        if package is not None:
582            result = JinjaMacroRegistry(packages={package: trimmed_macros})
583        else:
584            result = JinjaMacroRegistry(root_macros=trimmed_macros)
585
586        for upstream_package, upstream_names in dependencies.items():
587            result = result.merge(
588                self._trim_macros(upstream_names, upstream_package, visited=visited)
589            )
590
591        return result
592
593    def _macro_exists(self, name: str, package: t.Optional[str]) -> bool:
594        return (
595            name in self.packages.get(package, {})
596            if package is not None
597            else name in self.root_macros
598        )
599
600    def _get_macro(self, name: str, package: t.Optional[str]) -> MacroInfo:
601        return self.packages[package][name] if package is not None else self.root_macros[name]
602
603    def _to_non_private_macro_def(self, name: str, template: nodes.Template) -> nodes.Template:
604        for node in template.find_all((nodes.Macro, nodes.Call)):
605            if isinstance(node, nodes.Macro):
606                node.name = _non_private_name(name)
607            elif isinstance(node, nodes.Call) and isinstance(node.node, nodes.Name):
608                node.node.name = _non_private_name(name)
609
610        return template
611
612    def _create_builtin_globals(self, global_vars: t.Dict[str, t.Any]) -> t.Dict[str, t.Any]:
613        """Creates Jinja builtin globals using a factory function defined in the provided module."""
614        engine_adapter = global_vars.pop("engine_adapter", None)
615        global_vars = {**self.global_objs, **global_vars}
616        if self.create_builtins_module is not None:
617            module = importlib.import_module(self.create_builtins_module)
618            if hasattr(module, "create_builtin_globals"):
619                return module.create_builtin_globals(self, global_vars, engine_adapter)
620        return global_vars
621
622    def _create_builtin_filters(self) -> t.Dict[str, t.Any]:
623        """Creates Jinja builtin filters using a factory function defined in the provided module."""
624        if self.create_builtins_module is not None:
625            module = importlib.import_module(self.create_builtins_module)
626            if hasattr(module, "create_builtin_filters"):
627                return module.create_builtin_filters()
628        return {}
629
630    class _MacroWrapper:
631        def __init__(
632            self,
633            name: str,
634            package: t.Optional[str],
635            registry: JinjaMacroRegistry,
636            context: t.Dict[str, t.Any],
637        ):
638            self.name = name
639            self.package = package
640            self.context = context
641            self.registry = registry
642
643        def __call__(self, *args: t.Any, **kwargs: t.Any) -> t.Any:
644            context = self.context.copy()
645            if self.package is not None and self.package in context:
646                context.update(context[self.package])
647
648            template = self.registry._parse_macro(self.name, self.package)
649            macro_callable = getattr(
650                template.make_module(vars=context), _non_private_name(self.name)
651            )
652            try:
653                return macro_callable(*args, **kwargs)
654            except MacroReturnVal as ret:
655                return ret.value
656
657
658def _is_private_macro(name: str) -> bool:
659    return name.startswith("_")
660
661
662def _non_private_name(name: str) -> str:
663    return name.lstrip("_")
664
665
666JINJA_REGEX = re.compile(r"({{|{%)")
667
668
669def has_jinja(value: str) -> bool:
670    return JINJA_REGEX.search(value) is not None
671
672
673def jinja_call_arg_name(node: nodes.Node) -> str:
674    if isinstance(node, nodes.Const):
675        return node.value
676    return ""
677
678
679def create_var(variables: t.Dict[str, t.Any]) -> t.Callable:
680    def _var(var_name: str, default: t.Optional[t.Any] = None) -> t.Optional[t.Any]:
681        value = variables.get(var_name.lower(), default)
682        if isinstance(value, SqlValue):
683            return value.sql
684        return value
685
686    return _var
687
688
689def create_builtin_globals(
690    jinja_macros: JinjaMacroRegistry, global_vars: t.Dict[str, t.Any], *args: t.Any, **kwargs: t.Any
691) -> t.Dict[str, t.Any]:
692    global_vars.pop(c.GATEWAY, None)
693    variables = global_vars.pop(c.SQLMESH_VARS, None) or {}
694    blueprint_variables = global_vars.pop(c.SQLMESH_BLUEPRINT_VARS, None) or {}
695    return {
696        **global_vars,
697        c.VAR: create_var(variables),
698        c.GATEWAY: lambda: variables.get(c.GATEWAY, None),
699        c.BLUEPRINT_VAR: create_var(blueprint_variables),
700    }
701
702
703def make_jinja_registry(
704    jinja_macros: JinjaMacroRegistry, package_name: str, jinja_references: t.Set[MacroReference]
705) -> JinjaMacroRegistry:
706    """
707    Creates a Jinja macro registry for a specific package.
708
709    This function takes an existing Jinja macro registry and returns a new
710    registry that includes only the macros associated with the specified
711    package and trims the registry to include only the macros referenced
712    in the provided set of macro references.
713
714    Args:
715        jinja_macros: The original Jinja macro registry containing all macros.
716        package_name: The name of the package for which to create the registry.
717        jinja_references: A set of macro references to retain in the new registry.
718
719    Returns:
720        A new JinjaMacroRegistry containing only the macros for the specified
721        package and the referenced macros.
722    """
723
724    jinja_registry = jinja_macros.copy()
725    jinja_registry.root_macros = jinja_registry.packages.get(package_name) or {}
726    jinja_registry = jinja_registry.trim(jinja_references)
727
728    return jinja_registry
729
730
731def extract_error_details(ex: Exception) -> str:
732    """Extracts a readable message from a Jinja2 error, to include missing name and macro."""
733
734    error_details = ""
735    if isinstance(ex, UndefinedError):
736        if match := re.search(r"'(\w+)'", str(ex)):
737            error_details += f"\nUndefined macro/variable: '{match.group(1)}'"
738        try:
739            _, _, exc_traceback = exc_info()
740            for frame, _ in walk_tb(exc_traceback):
741                if frame.f_code.co_name == "_invoke":
742                    macro = frame.f_locals.get("self")
743                    if isinstance(macro, Macro):
744                        error_details += f" in macro: '{macro.name}'\n"
745                        break
746        except:
747            # to fall back to the generic error message if frame analysis fails
748            pass
749    return error_details or str(ex)