pyrtan / src / pyrtan / template.py
5,945 bytes · last changed 2026-10-28 in Add exchange rates and synonyms, with a join filter for lists in sentences
| 1 | """``{{ path }}`` interpolation against the run's state. |
| 2 | |
| 3 | Templates are deliberately tiny. A placeholder is a dotted path into the |
| 4 | state, optionally followed by one or more filters:: |
| 5 | |
| 6 | {{ inputs.city }} |
| 7 | {{ state.resp.json.results[0].name }} |
| 8 | {{ inputs.city | url }} |
| 9 | {{ state.count | int }} |
| 10 | |
| 11 | There is no logic in templates. Branching belongs to ``decide`` nodes, where |
| 12 | it is visible in the trace; a template that could branch would hide it. |
| 13 | """ |
| 14 | |
| 15 | from __future__ import annotations |
| 16 | |
| 17 | import json |
| 18 | import re |
| 19 | import urllib.parse |
| 20 | from typing import Any |
| 21 | |
| 22 | PLACEHOLDER = re.compile(r"\{\{\s*([^{}]*?)\s*\}\}") |
| 23 | PATH_TOKEN = re.compile(r"([A-Za-z_][A-Za-z0-9_\-]*)|\[(-?\d+)\]|\[\"([^\"]*)\"\]|\['([^']*)'\]") |
| 24 | |
| 25 | |
| 26 | class TemplateError(ValueError): |
| 27 | pass |
| 28 | |
| 29 | |
| 30 | class Missing: |
| 31 | """Sentinel for a path that does not resolve.""" |
| 32 | |
| 33 | def __repr__(self) -> str: |
| 34 | return "<missing>" |
| 35 | |
| 36 | |
| 37 | MISSING = Missing() |
| 38 | |
| 39 | |
| 40 | def resolve(path: str, scope: dict[str, Any]) -> Any: |
| 41 | """Resolve a dotted path like ``state.resp.json.items[0].title``. |
| 42 | |
| 43 | Returns :data:`MISSING` when any step does not exist, rather than raising, |
| 44 | so that ``exists`` conditions can be expressed. |
| 45 | """ |
| 46 | pos = 0 |
| 47 | cur: Any = scope |
| 48 | path = path.strip() |
| 49 | if not path: |
| 50 | return MISSING |
| 51 | while pos < len(path): |
| 52 | if path[pos] == ".": |
| 53 | pos += 1 |
| 54 | continue |
| 55 | m = PATH_TOKEN.match(path, pos) |
| 56 | if not m: |
| 57 | raise TemplateError(f"bad path {path!r} at offset {pos}") |
| 58 | pos = m.end() |
| 59 | name, index, key_dq, key_sq = m.groups() |
| 60 | if name is not None: |
| 61 | cur = _get_key(cur, name) |
| 62 | elif index is not None: |
| 63 | cur = _get_index(cur, int(index)) |
| 64 | else: |
| 65 | cur = _get_key(cur, key_dq if key_dq is not None else key_sq) |
| 66 | if cur is MISSING: |
| 67 | return MISSING |
| 68 | return cur |
| 69 | |
| 70 | |
| 71 | def _get_key(cur: Any, key: str) -> Any: |
| 72 | if isinstance(cur, dict): |
| 73 | return cur.get(key, MISSING) |
| 74 | if isinstance(cur, list) and key.isdigit(): |
| 75 | return _get_index(cur, int(key)) |
| 76 | return MISSING |
| 77 | |
| 78 | |
| 79 | def _get_index(cur: Any, index: int) -> Any: |
| 80 | if isinstance(cur, (list, tuple, str)): |
| 81 | try: |
| 82 | return cur[index] |
| 83 | except IndexError: |
| 84 | return MISSING |
| 85 | return MISSING |
| 86 | |
| 87 | |
| 88 | def _text(value: Any) -> str: |
| 89 | """A value as it reads inside a sentence: a list or table as JSON, the rest as itself.""" |
| 90 | if isinstance(value, (dict, list)): |
| 91 | return json.dumps(value, ensure_ascii=False) |
| 92 | return str(value) |
| 93 | |
| 94 | |
| 95 | FILTERS = { |
| 96 | "url": lambda v: urllib.parse.quote(str(v), safe=""), |
| 97 | "json": lambda v: json.dumps(v, ensure_ascii=False), |
| 98 | "int": lambda v: int(float(v)), |
| 99 | "float": lambda v: float(v), |
| 100 | "str": lambda v: str(v), |
| 101 | "lower": lambda v: str(v).lower(), |
| 102 | "upper": lambda v: str(v).upper(), |
| 103 | "strip": lambda v: str(v).strip(), |
| 104 | "first": lambda v: v[0] if isinstance(v, (list, tuple, str)) and v else MISSING, |
| 105 | "last": lambda v: v[-1] if isinstance(v, (list, tuple, str)) and v else MISSING, |
| 106 | "length": lambda v: len(v) if hasattr(v, "__len__") else MISSING, |
| 107 | "title": lambda v: str(v).title(), |
| 108 | # A list in a sentence: "a, b, c". Its items read as they would on their own. |
| 109 | "join": lambda v: ", ".join(_text(x) for x in v) if isinstance(v, (list, tuple)) else _text(v), |
| 110 | } |
| 111 | |
| 112 | |
| 113 | def render_value(expr: str, scope: dict[str, Any]) -> Any: |
| 114 | """Evaluate one placeholder expression (path plus filters) to a value.""" |
| 115 | parts = [p.strip() for p in expr.split("|")] |
| 116 | value = resolve(parts[0], scope) |
| 117 | for name in parts[1:]: |
| 118 | if name not in FILTERS: |
| 119 | raise TemplateError(f"unknown filter {name!r} in {{{{ {expr} }}}}") |
| 120 | if value is MISSING: |
| 121 | continue |
| 122 | value = FILTERS[name](value) |
| 123 | return value |
| 124 | |
| 125 | |
| 126 | def render(template: Any, scope: dict[str, Any], *, strict: bool = True) -> Any: |
| 127 | """Render a template. |
| 128 | |
| 129 | - A string that is exactly one placeholder returns the resolved value with |
| 130 | its type intact, so ``"{{ state.ids }}"`` yields a list, not its repr. |
| 131 | - A string with surrounding text interpolates placeholders as text. |
| 132 | - Lists and dicts are rendered recursively. |
| 133 | - Other values pass through. |
| 134 | |
| 135 | With ``strict`` (the default) a placeholder that resolves to nothing raises |
| 136 | :class:`TemplateError`; a tree should decide on absence explicitly rather |
| 137 | than paste an empty string into a URL. |
| 138 | """ |
| 139 | if isinstance(template, str): |
| 140 | whole = PLACEHOLDER.fullmatch(template.strip()) |
| 141 | if whole and template.strip() == template: |
| 142 | value = render_value(whole.group(1), scope) |
| 143 | if value is MISSING: |
| 144 | if strict: |
| 145 | raise TemplateError(f"{template} resolves to nothing") |
| 146 | return None |
| 147 | return value |
| 148 | |
| 149 | def sub(m: re.Match[str]) -> str: |
| 150 | value = render_value(m.group(1), scope) |
| 151 | if value is MISSING: |
| 152 | if strict: |
| 153 | raise TemplateError(f"{m.group(0)} resolves to nothing") |
| 154 | return "" |
| 155 | return _text(value) |
| 156 | |
| 157 | return PLACEHOLDER.sub(sub, template) |
| 158 | if isinstance(template, list): |
| 159 | return [render(item, scope, strict=strict) for item in template] |
| 160 | if isinstance(template, dict): |
| 161 | return {k: render(v, scope, strict=strict) for k, v in template.items()} |
| 162 | return template |
| 163 | |
| 164 | |
| 165 | def placeholders(template: Any) -> list[str]: |
| 166 | """Every placeholder path used anywhere in a template, filters stripped.""" |
| 167 | found: list[str] = [] |
| 168 | if isinstance(template, str): |
| 169 | for m in PLACEHOLDER.finditer(template): |
| 170 | found.append(m.group(1).split("|")[0].strip()) |
| 171 | elif isinstance(template, list): |
| 172 | for item in template: |
| 173 | found.extend(placeholders(item)) |
| 174 | elif isinstance(template, dict): |
| 175 | for v in template.values(): |
| 176 | found.extend(placeholders(v)) |
| 177 | return found |