feat(engine): Implement template expression functions (#154)

tests/unit/test_template_expressions.py

@@ -66,6 +66,10 @@ def test_eval_jsonpath():
         ("${{ INPUTS.arg2 -> str }}", "2"),
         ("${{ ACTIONS.webhook.result -> str }}", "1"),
         ("${{ ACTIONS.path_A_first.result.path.nested.value -> int }}", 9999),
+        (
+            "${{ FNS.add(INPUTS.arg1, ACTIONS.path_A_first.result.path.nested.value) }}",
+            10000,
+        ),
     ],
 )
 def test_templated_expression_result(expression, expected_result):
@@ -88,6 +92,47 @@ def test_templated_expression_result(expression, expected_result):
     assert fut.result() == expected_result
 
 
+@pytest.mark.parametrize(
+    "expression, expected_result",
+    [
+        (
+            "${{ FNS.is_equal(bool(True), bool(1)) -> bool }}",
+            True,
+        ),
+        (
+            "${{ FNS.add(int(1234), float(0.5)) -> float }}",
+            1234.5,
+        ),
+        (
+            "${{ FNS.less_than(INPUTS.arg1, INPUTS.arg2) -> bool }}",
+            True,
+        ),
+        (
+            "${{ FNS.is_equal(INPUTS.arg1, ACTIONS.webhook.result) -> bool }}",
+            True,
+        ),
+    ],
+)
+def test_templated_expression_function(expression, expected_result):
+    exec_vars = {
+        "INPUTS": {
+            "arg1": 1,
+            "arg2": 2,
+        },
+        "ACTIONS": {
+            "webhook": {"result": 1},
+            "path_A_first": {"result": {"path": {"nested": {"value": 9999}}}},
+            "path_A_second": {"result": 3},
+            "path_B_first": {"result": 4},
+            "path_B_second": {"result": 5},
+        },
+        "metadata": {"name": "John Doe", "age": 30},
+    }
+
+    fut = TemplateExpression(expression, operand=exec_vars)
+    assert fut.result() == expected_result
+
+
 def test_find_secrets():
     # Test for finding secrets in a string
     test_str = "This is a ${{ SECRETS.my_secret.TEST_API_KEY_1 }} secret"

tracecat/templates/expressions.py

@@ -39,6 +39,9 @@
 
 """
 
+import operator
+import re
+from collections.abc import Iterable
 from typing import Any, Literal, TypeVar
 
 import jsonpath_ng
@@ -47,23 +50,58 @@
 from tracecat.templates import patterns
 
 T = TypeVar("T")
-token_pattern = r""
+
+
+def _bool(x: Any) -> bool:
+    if isinstance(x, bool):
+        return x
+    if isinstance(x, str):
+        return x.lower() in ("true", "1")
+    # Everything else is False
+    return False
+
+
 _BUILTIN_TYPE_NAP = {
     "int": int,
     "float": float,
     "str": str,
+    "bool": _bool,
     # TODO: Perhaps support for URLs for files?
 }
-TemplateStr = str
+ExprStr = str
 """An annotated template string that can be resolved into a type T."""
 
 OperandType = dict[str, Any]
+_FN_TABLE = {
+    # Comparison
+    "less_than": operator.lt,
+    "less_than_or_equal": operator.le,
+    "greater_than": operator.gt,
+    "greater_than_or_equal": operator.ge,
+    "not_equal": operator.ne,
+    "is_equal": operator.eq,
+    # Regex
+    "regex_match": lambda pattern, text: bool(re.match(pattern, text)),
+    "regex_not_match": lambda pattern, text: not bool(re.match(pattern, text)),
+    # Membership
+    "contains": lambda item, container: item in container,
+    "does_not_contain": lambda item, container: item not in container,
+    # Math
+    "add": operator.add,
+    "sub": operator.sub,
+    "mul": operator.mul,
+    "div": operator.truediv,
+    "mod": operator.mod,
+    "pow": operator.pow,
+}
 
 
-class TemplateExpression:
-    """A expression that resolves into a type T."""
+class Expression:
+    """An expression.
+    i.e. `${{ <expression> -> <type> }}`
+    """
 
-    _context: Literal["SECRETS", "FNS", "ACTIONS", "INPUTS"]
+    _context: Literal["SECRETS", "FNS", "ACTIONS", "INPUTS", "ENV"]
     """The context of the expression, e.g. SECRETS, FNS, ACTIONS, INPUTS, or a jsonpath."""
 
     _expr: str
@@ -72,17 +110,17 @@ class TemplateExpression:
     _operand: OperandType | None
 
     def __init__(
-        self, template: TemplateStr, *, operand: OperandType | None = None
+        self, expression: ExprStr, *, operand: OperandType | None = None
     ) -> None:
-        self._template = template
-        match = patterns.TYPED_TEMPLATE.match(template)
+        self._template = expression
+        match = patterns.EXPR_PATTERN.match(expression)
         if not match:
-            raise ValueError(f"Invalid template: {template!r}")
+            raise ValueError(f"Invalid expression: {expression!r}")
 
         # Top level types
         self._context = match.group("context")
         self._expr = match.group("expr")
-        self._resolve_typename = match.group("type")
+        self._resolve_typename = match.group("rtype")
         # If no type is specified, do not cast the result
         self._resolve_type = _BUILTIN_TYPE_NAP.get(self._resolve_typename)
 
@@ -103,16 +141,24 @@ def __repr__(self) -> str:
         )
 
     def _resolve_fn(self) -> Any:
-        return NotImplemented
-        # matched_fn = patterns.EXPR_INLINE_FN.match(self._expr)
-        # if not matched_fn:
-        #     raise ValueError(f"Invalid function expression: {self._expr!r}")
-        # print("Got an inline function")
-        # fn_name = matched_fn.group("func")
-        # fn_args = re.split(r",\s*", matched_fn.group("args"))
-        # # Get the function, likely from some regsitry or module and call it
-        # print(fn_name, fn_args)
-        # return fn_name
+        """Resolve a funciton expression."""
+
+        matched_fn = patterns.EXPR_INLINE_FN.match(self._expr)
+        if not matched_fn:
+            raise ValueError(f"Invalid function expression: {self._expr!r}")
+        fn_name = matched_fn.group("func")
+        fn_args = re.split(r",\s*", matched_fn.group("args"))
+
+        # Resolve all args into the correct type
+        resolved_args = self._evaluate_function_args(fn_args)
+        result = _FN_TABLE[fn_name](*resolved_args)
+
+        return result
+
+    def _evaluate_function_args(self, args: Iterable[str]) -> tuple[str]:
+        """Evaluate function args inside a template expression"""
+
+        return tuple(eval_inline_expression(arg, self._operand) for arg in args)
 
     def result(self) -> Any:
         """Evaluate the templated expression and return the result."""
@@ -133,6 +179,26 @@ def result(self) -> Any:
             raise ValueError(f"Could not cast {ret!r} to {self._resolve_type!r}") from e
 
 
+class TemplateExpression:
+    """Expression with template syntax."""
+
+    expr: Expression
+
+    def __init__(
+        self,
+        template: str,
+        operand: OperandType | None = None,
+        pattern: re.Pattern[str] = patterns.TEMPLATED_OBJ,
+    ) -> None:
+        match = pattern.match(template)
+        if (expr := match.group("expr")) is None:
+            raise ValueError(f"Invalid template expression: {template!r}")
+        self.expr = Expression(expr, operand=operand)
+
+    def result(self) -> Any:
+        return self.expr.result()
+
+
 def eval_jsonpath(expr: str, operand: dict[str, Any]) -> Any:
     if operand is None or not isinstance(operand, dict):
         raise ValueError("A dict-type operand is required for templated jsonpath.")
@@ -150,3 +216,21 @@ def eval_jsonpath(expr: str, operand: dict[str, Any]) -> Any:
         # We know that if this function is called, there was a templated field.
         # Therefore, it means the jsonpath was valid but there was no match.
         raise ValueError(f"Operand has no path {expr!r}. Operand: {operand}.")
+
+
+def eval_inline_expression(expr: str, operand: OperandType) -> Any:
+    """Evaluate inline expressions like
+    - Expression: (with context) e.g. 'INPUTS.result', 'ACTIONS.step2.result'
+    - Inline typecast: values like e.g. 'int(5)', 'str("hello")'
+    """
+
+    if match := patterns.EXPR_PATTERN.match(expr):
+        full_expr = match.group("full")
+        return Expression(full_expr, operand=operand).result()
+
+    if match := patterns.INLINE_TYPECAST.match(expr):
+        type_name = match.group("type")
+        value = match.group("value")
+        return _BUILTIN_TYPE_NAP[type_name](value)
+
+    raise ValueError(f"Invalid function argument: {expr!r}")

tracecat/templates/patterns.py

@@ -1,18 +1,34 @@
 import re
 
-TEMPLATED_OBJ = re.compile(r"(?P<template>\${{\s*.+?\s*}})")  # Lazy match
+TEMPLATED_OBJ = re.compile(r"(?P<template>\${{\s(?P<expr>.+?)\s*}})")  # Lazy match
+
+EXPR_PATTERN = re.compile(
+    r"""
+    ^\s*                                                    #  Optional whitespace
+    (?P<full>                                               # Capture the full expression
+    (?P<context>INPUTS|ACTIONS|SECRETS|FNS|ENV?)            # Non-greedy capture for 'expression', any chars
+    \.
+    (?P<expr>.+?)            # Non-greedy capture for 'expression', any chars
+    (\s*->\s*(?P<rtype>int|float|str|bool))?
+    )                          # Capture 'type', which must be one of 'int', 'float', 'str'
+    (?=\s*$)                                               # Optional whitespace
+    """,
+    re.VERBOSE,
+)
 
 TYPED_TEMPLATE = re.compile(
     r"""
     \${{\s*                                               # Opening curly braces and optional whitespace
-    (?P<context>INPUTS|ACTIONS|SECRETS|FNS?)            # Non-greedy capture for 'expression', any chars
+    (?P<context>INPUTS|ACTIONS|SECRETS|FNS|ENV?)            # Non-greedy capture for 'expression', any chars
     \.
     (?P<expr>.+?)            # Non-greedy capture for 'expression', any chars
-    (\s*->\s*(?P<type>int|float|str))?                             # Capture 'type', which must be one of 'int', 'float', 'str'
+    (\s*->\s*(?P<rtype>int|float|str|bool))?                             # Capture 'type', which must be one of 'int', 'float', 'str'
     \s*}}                                               # Optional whitespace and closing curly braces
 """,
     re.VERBOSE,
 )
+
+
 SECRET_TEMPLATE = re.compile(
     r"""
     \${{\s*                                               # Opening curly braces and optional whitespace
@@ -47,3 +63,14 @@
 EXPR_QUALIFIED_ATTRIBUTE = re.compile(r"\b[a-zA-Z0-9]+(?:\.[a-zA-Z0-9]+)+\b")
 
 FULL_TEMPLATE = re.compile(r"^\${{\s*[^{}]*\s*}}$")
+
+# Match "int(5)" or "float(5.0)" or "str('hello')" or "bool(True)"
+INLINE_TYPECAST = re.compile(
+    r"""
+    (?P<type>int|float|str|bool)  # Match one of 'int', 'float', 'str', 'bool'
+    \(
+    (?P<value>.+?)                # Non-greedy capture for 'value', any characters
+    \)
+""",
+    re.VERBOSE,
+)