generated from coulomb/repo-seed
Bernd Worsch
bac3efee89
activities.py — resolve_context (T15):
- dispatches on source.type: 'static' returns config["value"]
- 'http_get' / 'db_query' raise ApplicationError(non_retryable=True)
- unknown types raise ApplicationError(non_retryable=True)
template_engine.py — evaluate_templates (T16, pure function):
- evaluates optional condition expressions against context snapshot
(restricted eval, no builtins)
- interpolates {context.<name>.<key>} placeholders via str.format_map
- returns list[{task_type, params}] with falsy-condition rows omitted
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
102 lines
3.3 KiB
Python
102 lines
3.3 KiB
Python
"""Pure-function template evaluation for activity-core.
|
|
|
|
evaluate_templates() is called directly inside RunActivityWorkflow (not as
|
|
an activity) because it is purely deterministic and has no I/O.
|
|
|
|
Template interpolation syntax:
|
|
params_template values may contain {context.<name>.<key>} placeholders,
|
|
resolved against the context snapshot via Python's str.format_map.
|
|
|
|
Example:
|
|
context_snapshot = {"stub": {"ping": "pong"}}
|
|
params_template = {"message": "hello {context.stub.ping}"}
|
|
result = {"message": "hello pong"}
|
|
|
|
Condition syntax:
|
|
A Python expression string evaluated against {"context": context_snapshot}.
|
|
The task is included only when the expression is truthy.
|
|
Builtins are intentionally stripped — keep conditions simple comparisons.
|
|
|
|
Example: "context['user']['is_active'] == True"
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
|
|
|
|
class _DotAccessDict(dict):
|
|
"""Dict subclass that allows attribute-style access for str.format_map.
|
|
|
|
Enables {context.name.key} in format strings by making nested dicts
|
|
accessible as attributes.
|
|
"""
|
|
|
|
def __getattr__(self, name: str) -> object:
|
|
try:
|
|
value = self[name]
|
|
except KeyError:
|
|
raise AttributeError(name) from None
|
|
if isinstance(value, dict):
|
|
return _DotAccessDict(value)
|
|
return value
|
|
|
|
|
|
def _interpolate(value: object, context_snapshot: dict) -> object:
|
|
"""Recursively interpolate {context.*} placeholders in a params value."""
|
|
if isinstance(value, str):
|
|
return value.format_map({"context": _DotAccessDict(context_snapshot)})
|
|
if isinstance(value, dict):
|
|
return {k: _interpolate(v, context_snapshot) for k, v in value.items()}
|
|
if isinstance(value, list):
|
|
return [_interpolate(item, context_snapshot) for item in value]
|
|
return value
|
|
|
|
|
|
def _eval_condition(condition: str, context_snapshot: dict) -> bool:
|
|
"""Evaluate a condition expression against the context snapshot.
|
|
|
|
Raises:
|
|
ValueError: if the expression raises or returns a non-bool-able value.
|
|
"""
|
|
try:
|
|
result = eval( # noqa: S307
|
|
condition,
|
|
{"__builtins__": {}},
|
|
{"context": context_snapshot},
|
|
)
|
|
except Exception as exc:
|
|
raise ValueError(f"Condition {condition!r} raised: {exc}") from exc
|
|
return bool(result)
|
|
|
|
|
|
def evaluate_templates(
|
|
task_templates: list[dict],
|
|
context_snapshot: dict,
|
|
) -> list[dict]:
|
|
"""Expand task templates against a context snapshot.
|
|
|
|
Args:
|
|
task_templates: list of TaskTemplate dicts (task_type, condition,
|
|
params_template).
|
|
context_snapshot: merged context dict produced by resolve_context.
|
|
|
|
Returns:
|
|
List of concrete task specs: [{"task_type": str, "params": dict}, ...]
|
|
Templates whose condition evaluates falsy are omitted.
|
|
"""
|
|
results: list[dict] = []
|
|
|
|
for template in task_templates:
|
|
condition = template.get("condition")
|
|
if condition is not None and not _eval_condition(condition, context_snapshot):
|
|
continue
|
|
|
|
params = _interpolate(
|
|
copy.deepcopy(template.get("params_template", {})),
|
|
context_snapshot,
|
|
)
|
|
results.append({"task_type": template["task_type"], "params": params})
|
|
|
|
return results
|