2025-12-22 14:08:30 +05:30
|
|
|
|
"""Template rendering utility with support for nested JSON paths."""
|
2025-09-09 14:37:32 +05:30
|
|
|
|
|
2025-12-22 14:08:30 +05:30
|
|
|
|
import json
|
2025-09-09 14:37:32 +05:30
|
|
|
|
import re
|
2025-12-22 14:08:30 +05:30
|
|
|
|
from typing import Any, Dict, Union
|
2025-09-09 14:37:32 +05:30
|
|
|
|
|
2026-03-19 17:52:04 +05:30
|
|
|
|
from api.services.workflow.workflow import TEMPLATE_VAR_PATTERN
|
|
|
|
|
|
|
2025-09-09 14:37:32 +05:30
|
|
|
|
|
2025-12-22 14:08:30 +05:30
|
|
|
|
def get_nested_value(obj: Any, path: str) -> Any:
|
|
|
|
|
|
"""
|
|
|
|
|
|
Get a nested value from a dictionary using dot notation.
|
|
|
|
|
|
|
|
|
|
|
|
Args:
|
|
|
|
|
|
obj: The object to traverse (dict or any)
|
|
|
|
|
|
path: Dot-separated path (e.g., "a.b.c")
|
|
|
|
|
|
|
|
|
|
|
|
Returns:
|
|
|
|
|
|
The value at the path, or None if not found
|
|
|
|
|
|
|
|
|
|
|
|
Examples:
|
|
|
|
|
|
get_nested_value({"a": {"b": 1}}, "a.b") -> 1
|
|
|
|
|
|
get_nested_value({"a": {"b": {"c": 2}}}, "a.b.c") -> 2
|
|
|
|
|
|
get_nested_value({"a": 1}, "a.b") -> None
|
|
|
|
|
|
"""
|
|
|
|
|
|
if not path:
|
|
|
|
|
|
return obj
|
|
|
|
|
|
|
|
|
|
|
|
keys = path.split(".")
|
|
|
|
|
|
current = obj
|
|
|
|
|
|
|
|
|
|
|
|
for key in keys:
|
|
|
|
|
|
if isinstance(current, dict):
|
|
|
|
|
|
current = current.get(key)
|
|
|
|
|
|
else:
|
|
|
|
|
|
return None
|
|
|
|
|
|
|
|
|
|
|
|
if current is None:
|
|
|
|
|
|
return None
|
|
|
|
|
|
|
|
|
|
|
|
return current
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def render_template(
|
|
|
|
|
|
template: Union[str, dict, list, None],
|
|
|
|
|
|
context: Dict[str, Any],
|
|
|
|
|
|
) -> Union[str, dict, list, None]: # noqa: C901 – complex but self-contained
|
|
|
|
|
|
"""
|
|
|
|
|
|
Render a template with variable substitution supporting nested paths.
|
|
|
|
|
|
|
|
|
|
|
|
Supports:
|
|
|
|
|
|
- String templates: "Hello {{name}}"
|
|
|
|
|
|
- JSON templates: {"key": "{{value}}"}
|
|
|
|
|
|
- Nested paths: "{{initial_context.phone_number}}"
|
|
|
|
|
|
- Deep nesting: "{{gathered_context.customer.address.city}}"
|
|
|
|
|
|
- Fallback: "{{name | fallback:Unknown}}"
|
2025-09-09 14:37:32 +05:30
|
|
|
|
|
2025-12-22 14:08:30 +05:30
|
|
|
|
Args:
|
|
|
|
|
|
template: String, dict, list, or None with {{variable}} placeholders
|
|
|
|
|
|
context: Dict containing all available variables
|
2025-09-09 14:37:32 +05:30
|
|
|
|
|
2025-12-22 14:08:30 +05:30
|
|
|
|
Returns:
|
|
|
|
|
|
Rendered template with variables replaced
|
2025-09-09 14:37:32 +05:30
|
|
|
|
"""
|
2025-12-22 14:08:30 +05:30
|
|
|
|
if template is None:
|
|
|
|
|
|
return None
|
|
|
|
|
|
|
|
|
|
|
|
# Handle dict templates recursively
|
|
|
|
|
|
if isinstance(template, dict):
|
|
|
|
|
|
return {
|
|
|
|
|
|
_render_string(str(k), context)
|
|
|
|
|
|
if isinstance(k, str)
|
|
|
|
|
|
else k: render_template(v, context)
|
|
|
|
|
|
for k, v in template.items()
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
# Handle list templates recursively
|
|
|
|
|
|
if isinstance(template, list):
|
|
|
|
|
|
return [render_template(item, context) for item in template]
|
|
|
|
|
|
|
|
|
|
|
|
# Handle non-string types (int, float, bool, etc.)
|
|
|
|
|
|
if not isinstance(template, str):
|
|
|
|
|
|
return template
|
|
|
|
|
|
|
|
|
|
|
|
return _render_string(template, context)
|
2025-09-09 14:37:32 +05:30
|
|
|
|
|
2025-12-22 14:08:30 +05:30
|
|
|
|
|
|
|
|
|
|
def _render_string(template_str: str, context: Dict[str, Any]) -> str:
|
|
|
|
|
|
"""
|
|
|
|
|
|
Render a string template with variable substitution.
|
|
|
|
|
|
|
|
|
|
|
|
Args:
|
|
|
|
|
|
template_str: String with {{variable}} placeholders
|
|
|
|
|
|
context: Dict containing all available variables
|
|
|
|
|
|
|
|
|
|
|
|
Returns:
|
|
|
|
|
|
Rendered string with variables replaced
|
|
|
|
|
|
"""
|
2025-09-09 14:37:32 +05:30
|
|
|
|
if not template_str:
|
|
|
|
|
|
return template_str
|
|
|
|
|
|
|
|
|
|
|
|
def _replace(match: re.Match[str]) -> str: # type: ignore[type-arg]
|
2025-12-22 14:08:30 +05:30
|
|
|
|
variable_path = match.group(1).strip()
|
2025-09-09 14:37:32 +05:30
|
|
|
|
filter_name = match.group(2).strip() if match.group(2) else None
|
|
|
|
|
|
filter_value = match.group(3).strip() if match.group(3) else None
|
|
|
|
|
|
|
2025-12-22 14:08:30 +05:30
|
|
|
|
# Get value using nested path lookup
|
|
|
|
|
|
value = get_nested_value(context, variable_path)
|
2025-09-09 14:37:32 +05:30
|
|
|
|
|
|
|
|
|
|
# Apply filters
|
|
|
|
|
|
if filter_name == "fallback":
|
|
|
|
|
|
if value is None or value == "":
|
|
|
|
|
|
value = (
|
2025-12-22 14:08:30 +05:30
|
|
|
|
filter_value if filter_value is not None else variable_path.title()
|
2025-09-09 14:37:32 +05:30
|
|
|
|
)
|
|
|
|
|
|
|
2025-12-22 14:08:30 +05:30
|
|
|
|
# Convert to string for substitution
|
|
|
|
|
|
if value is None:
|
|
|
|
|
|
return ""
|
|
|
|
|
|
if isinstance(value, (dict, list)):
|
|
|
|
|
|
return json.dumps(value)
|
|
|
|
|
|
return str(value)
|
2025-09-09 14:37:32 +05:30
|
|
|
|
|
|
|
|
|
|
# Replace template variables
|
2026-03-19 17:52:04 +05:30
|
|
|
|
result = re.sub(TEMPLATE_VAR_PATTERN, _replace, template_str)
|
2025-09-09 14:37:32 +05:30
|
|
|
|
|
|
|
|
|
|
# Handle line breaks (convert literal \n to actual newlines)
|
|
|
|
|
|
result = result.replace("\\n", "\n")
|
|
|
|
|
|
|
|
|
|
|
|
return result
|