Skip to content

yaml_workflow.template

yaml_workflow.template

Template engine implementation using Jinja2.

Classes

AttrDict

Bases: dict

A dictionary that allows attribute access to its keys.

Source code in src/yaml_workflow/template.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
class AttrDict(dict):
    """A dictionary that allows attribute access to its keys."""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        for k, v in list(super().items()):
            if isinstance(v, dict) and not isinstance(v, AttrDict):
                self[k] = AttrDict(v)
            elif isinstance(v, (list, tuple)):
                self[k] = [AttrDict(i) if isinstance(i, dict) else i for i in v]

    def __getattr__(self, key: str) -> Any:
        try:
            # Always check the dictionary first
            if key in self:
                return self[key]
            # If the key doesn't exist, try to get it as a method
            if key in dir(dict):
                method = getattr(super(), key)
                # If it's a callable method, call it immediately
                if callable(method):
                    result = method()
                    return result
                return method
            raise KeyError(key)
        except KeyError as e:
            raise AttributeError(key)

    def __setattr__(self, key: str, value: Any) -> None:
        self[key] = value

    def items(self):
        """Override items to ensure it returns a list of tuples."""
        return list(super().items())
Functions
items()

Override items to ensure it returns a list of tuples.

Source code in src/yaml_workflow/template.py 
44
45
46
def items(self):
    """Override items to ensure it returns a list of tuples."""
    return list(super().items())

TemplateEngine

Template engine for processing Jinja2 templates.

Source code in src/yaml_workflow/template.py 
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
class TemplateEngine:
    """Template engine for processing Jinja2 templates."""

    def __init__(self):
        """Initialize the template engine with strict undefined behavior."""
        # Default environment without a loader
        self.env = Environment(
            # loader=None, # Explicitly no loader by default
            undefined=StrictUndefined,
            autoescape=False,
            trim_blocks=True,
            lstrip_blocks=True,
        )

    def _extract_variable_path(self, template_str: str, error_msg: str) -> str:
        """Extract the full variable path from the template string.

        Args:
            template_str: The template string being processed
            error_msg: The error message from Jinja2

        Returns:
            str: The full variable path
        """
        # Extract the undefined variable name from the error message
        if "'is undefined'" in error_msg:
            var_name = error_msg.split("'")[1]
        else:
            # Handle attribute error
            var_parts = error_msg.split("'")
            if len(var_parts) >= 2:
                var_name = var_parts[-2]
            else:
                var_name = "unknown"

        # Find the variable in the template string
        pattern = r"{{\s*([^}]+)\s*}}"
        matches = re.findall(pattern, template_str)
        for match in matches:
            if var_name in match:
                return match.strip()
        return var_name

    def process_template(
        self,
        template_str: str,
        variables: Optional[Dict[str, Any]] = None,
        searchpath: Optional[str] = None,
    ) -> Any:
        """Process a template string with the given variables.

        Args:
            template_str (str): The template string to process.
            variables (Optional[Dict[str, Any]], optional): Variables to use in template processing.
                Defaults to None.
            searchpath (Optional[str], optional): Filesystem path for includes/extends. Defaults to None.

        Returns:
            Any: The processed template value, preserving the original type.

        Raises:
            TemplateError: If there is an error processing the template.
        """
        try:
            # Initialize variables to empty dict if None
            vars_dict: Dict[str, Any] = variables if variables is not None else {}

            # Choose environment: Create a new one if searchpath is provided
            if searchpath:
                env = Environment(
                    loader=FileSystemLoader(searchpath=searchpath),
                    undefined=StrictUndefined,
                    autoescape=False,
                    trim_blocks=True,
                    lstrip_blocks=True,
                )
            else:
                env = self.env  # Use default env if no searchpath

            # If the template is just a variable reference, try to return the raw value
            if template_str.strip().startswith("{{") and template_str.strip().endswith(
                "}}"
            ):
                var_path = template_str.strip()[2:-2].strip()
                if "." in var_path:
                    parts = var_path.split(".")
                    current: Optional[Dict[str, Any]] = vars_dict
                    for part in parts:
                        if current is None or not isinstance(current, dict):
                            break
                        current = current.get(part)  # type: ignore
                    if current is not None:
                        return current

            # Create a template using the chosen environment
            template = env.from_string(template_str)

            # Convert variables to AttrDict for proper attribute access
            context = AttrDict(vars_dict)

            # Process the template with the wrapped variables
            return template.render(**context)

        except UndefinedError as e:
            # Get the full variable path from the template
            var_path = self._extract_variable_path(template_str, str(e))
            parts = var_path.split(".")

            # Handle invalid namespace
            if len(parts) > 0:
                namespace = parts[0]
                if namespace not in vars_dict:
                    error_msg = (
                        f"Template error: Invalid namespace '{namespace}'\n"
                        f"Available namespaces:\n"
                    )
                    for ns in sorted(vars_dict.keys()):
                        if isinstance(vars_dict[ns], dict):
                            error_msg += f"  - {ns}\n"
                    raise TemplateError(error_msg)

                # Handle invalid attribute access
                if len(parts) > 2:
                    try:
                        current = vars_dict[namespace]
                        if not isinstance(current, dict):
                            raise TemplateError(
                                f"Template error: Cannot access attributes of non-dictionary value '{namespace}'"
                            )
                        for part in parts[1:-1]:
                            if not isinstance(current, dict):
                                raise TemplateError(
                                    f"Template error: Cannot access attributes of non-dictionary value '{'.'.join(parts[:-1])}'"
                                )
                            current = current[part]
                        error_msg = (
                            f"Template error: Invalid attribute '{parts[-1]}' on {type(current).__name__}\n"
                            f"Type of '{'.'.join(parts[:-1])}' is '{type(current).__name__}'"
                        )
                        raise TemplateError(error_msg)
                    except (KeyError, AttributeError):
                        pass

                # Handle undefined variable in namespace
                error_msg = (
                    f"Template error: Undefined variable '{var_path}'\n"
                    f"Available variables in '{namespace}' namespace:\n"
                )
                if namespace in vars_dict and isinstance(vars_dict[namespace], dict):
                    for key in sorted(vars_dict[namespace].keys()):
                        error_msg += f"  - {key}\n"
                raise TemplateError(error_msg)

            # Handle root level undefined variable
            error_msg = f"Template error: Undefined variable '{var_path}'\n"
            if vars_dict:
                error_msg += "Available variables:\n"
                for key in sorted(vars_dict.keys()):
                    error_msg += f"  - {key}\n"
            raise TemplateError(error_msg)

        except TemplateSyntaxError as e:
            raise TemplateError(f"Template syntax error: {str(e)}")
        except Exception as e:
            raise TemplateError(f"Error processing template: {str(e)}")

    def process_value(self, value: Any, variables: Dict[str, Any]) -> Any:
        """Process a value that may contain templates.

        Args:
            value: The value to process
            variables: Dictionary of variables to use in template processing

        Returns:
            Any: The processed value
        """
        if isinstance(value, str):
            return self.process_template(value, variables)
        elif isinstance(value, dict):
            return {k: self.process_value(v, variables) for k, v in value.items()}
        elif isinstance(value, list):
            return [self.process_value(item, variables) for item in value]
        return value
Functions
process_template(template_str: str, variables: Optional[Dict[str, Any]] = None, searchpath: Optional[str] = None) -> Any

Process a template string with the given variables.

Parameters:

Name Type Description Default
template_str str

The template string to process.

 required
variables Optional[Dict[str, Any]]

Variables to use in template processing. Defaults to None.

 None
searchpath Optional[str]

Filesystem path for includes/extends. Defaults to None.

 None

Returns:

Name Type Description
Any Any

The processed template value, preserving the original type.

Raises:

Type Description
TemplateError

If there is an error processing the template.
Source code in src/yaml_workflow/template.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def process_template(
    self,
    template_str: str,
    variables: Optional[Dict[str, Any]] = None,
    searchpath: Optional[str] = None,
) -> Any:
    """Process a template string with the given variables.

    Args:
        template_str (str): The template string to process.
        variables (Optional[Dict[str, Any]], optional): Variables to use in template processing.
            Defaults to None.
        searchpath (Optional[str], optional): Filesystem path for includes/extends. Defaults to None.

    Returns:
        Any: The processed template value, preserving the original type.

    Raises:
        TemplateError: If there is an error processing the template.
    """
    try:
        # Initialize variables to empty dict if None
        vars_dict: Dict[str, Any] = variables if variables is not None else {}

        # Choose environment: Create a new one if searchpath is provided
        if searchpath:
            env = Environment(
                loader=FileSystemLoader(searchpath=searchpath),
                undefined=StrictUndefined,
                autoescape=False,
                trim_blocks=True,
                lstrip_blocks=True,
            )
        else:
            env = self.env  # Use default env if no searchpath

        # If the template is just a variable reference, try to return the raw value
        if template_str.strip().startswith("{{") and template_str.strip().endswith(
            "}}"
        ):
            var_path = template_str.strip()[2:-2].strip()
            if "." in var_path:
                parts = var_path.split(".")
                current: Optional[Dict[str, Any]] = vars_dict
                for part in parts:
                    if current is None or not isinstance(current, dict):
                        break
                    current = current.get(part)  # type: ignore
                if current is not None:
                    return current

        # Create a template using the chosen environment
        template = env.from_string(template_str)

        # Convert variables to AttrDict for proper attribute access
        context = AttrDict(vars_dict)

        # Process the template with the wrapped variables
        return template.render(**context)

    except UndefinedError as e:
        # Get the full variable path from the template
        var_path = self._extract_variable_path(template_str, str(e))
        parts = var_path.split(".")

        # Handle invalid namespace
        if len(parts) > 0:
            namespace = parts[0]
            if namespace not in vars_dict:
                error_msg = (
                    f"Template error: Invalid namespace '{namespace}'\n"
                    f"Available namespaces:\n"
                )
                for ns in sorted(vars_dict.keys()):
                    if isinstance(vars_dict[ns], dict):
                        error_msg += f"  - {ns}\n"
                raise TemplateError(error_msg)

            # Handle invalid attribute access
            if len(parts) > 2:
                try:
                    current = vars_dict[namespace]
                    if not isinstance(current, dict):
                        raise TemplateError(
                            f"Template error: Cannot access attributes of non-dictionary value '{namespace}'"
                        )
                    for part in parts[1:-1]:
                        if not isinstance(current, dict):
                            raise TemplateError(
                                f"Template error: Cannot access attributes of non-dictionary value '{'.'.join(parts[:-1])}'"
                            )
                        current = current[part]
                    error_msg = (
                        f"Template error: Invalid attribute '{parts[-1]}' on {type(current).__name__}\n"
                        f"Type of '{'.'.join(parts[:-1])}' is '{type(current).__name__}'"
                    )
                    raise TemplateError(error_msg)
                except (KeyError, AttributeError):
                    pass

            # Handle undefined variable in namespace
            error_msg = (
                f"Template error: Undefined variable '{var_path}'\n"
                f"Available variables in '{namespace}' namespace:\n"
            )
            if namespace in vars_dict and isinstance(vars_dict[namespace], dict):
                for key in sorted(vars_dict[namespace].keys()):
                    error_msg += f"  - {key}\n"
            raise TemplateError(error_msg)

        # Handle root level undefined variable
        error_msg = f"Template error: Undefined variable '{var_path}'\n"
        if vars_dict:
            error_msg += "Available variables:\n"
            for key in sorted(vars_dict.keys()):
                error_msg += f"  - {key}\n"
        raise TemplateError(error_msg)

    except TemplateSyntaxError as e:
        raise TemplateError(f"Template syntax error: {str(e)}")
    except Exception as e:
        raise TemplateError(f"Error processing template: {str(e)}")
process_value(value: Any, variables: Dict[str, Any]) -> Any

Process a value that may contain templates.

Parameters:

Name Type Description Default
value Any

The value to process

 required
variables Dict[str, Any]

Dictionary of variables to use in template processing

 required

Returns:

Name Type Description
Any Any

The processed value
Source code in src/yaml_workflow/template.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
def process_value(self, value: Any, variables: Dict[str, Any]) -> Any:
    """Process a value that may contain templates.

    Args:
        value: The value to process
        variables: Dictionary of variables to use in template processing

    Returns:
        Any: The processed value
    """
    if isinstance(value, str):
        return self.process_template(value, variables)
    elif isinstance(value, dict):
        return {k: self.process_value(v, variables) for k, v in value.items()}
    elif isinstance(value, list):
        return [self.process_value(item, variables) for item in value]
    return value