condition builder

tests/test_demo_workflow.py

@@ -16,7 +16,11 @@ from wf_core.run_factory import create_run_state
 from wf_authoring import (
     NodeReturn,
     WorkflowBuilder,
+    bind_fields,
+    bind_state,
     build_registry,
+    exists,
+    state,
     context_path,
     input_path,
     node,
@@ -170,8 +174,8 @@ def build_authoring_demo_workflow():
     list_files = builder.use(
         drive_list_files_spec,
         id="list_files",
-        in_map={input_path("folder_id"): "folder_id"},
-        out_map={"documents": state_path("documents")},
+        in_map=bind_fields(folder_id=input_path("folder_id")),
+        out_map=bind_state(documents=state_path("documents")),
         desc="List files from a Google Drive folder",
     )
     summarize_each = builder.foreach(
@@ -184,49 +188,45 @@ def build_authoring_demo_workflow():
     summarize_one = builder.use(
         summarize_document_spec,
         id="summarize_one",
-        in_map={context_path("document"): "document"},
-        out_map={"item_summary": state_path("item_summaries")},
+        in_map=bind_fields(document=context_path("document")),
+        out_map=bind_state(item_summary=state_path("item_summaries")),
         desc="Summarize one document",
     )
     combine_summaries = builder.use(
         combine_summaries_spec,
         id="combine_summaries",
-        in_map={state_path("item_summaries"): "item_summaries"},
-        out_map={"summary": state_path("summary")},
+        in_map=bind_fields(item_summaries=state_path("item_summaries")),
+        out_map=bind_state(summary=state_path("summary")),
         desc="Combine item summaries into one final summary",
     )
     should_email = builder.condition(
         id="should_email",
-        check={
-            "op": "eq",
-            "left": {"path": "state.should_email"},
-            "right": {"value": True},
-        },
+        check=state("should_email").eq(True),
     )
     send_email = builder.use(
         send_email_spec,
         id="send_email",
-        in_map={state_path("summary"): "summary"},
-        out_map={"email_status": state_path("email_status")},
+        in_map=bind_fields(summary=state_path("summary")),
+        out_map=bind_state(email_status=state_path("email_status")),
         desc="Send the summary by email",
     )
     approve_email = builder.interrupt(
         id="approve_email",
         kind="approval",
-        request_map={
-            state_path("summary"): "summary",
-            input_path("folder_id"): "folder_id",
-        },
-        out_map={
-            "approved": state_path("approved"),
-            "comment": state_path("approval_comment"),
-        },
+        request_map=bind_fields(
+            summary=state_path("summary"),
+            folder_id=input_path("folder_id"),
+        ),
+        out_map=bind_state(
+            approved=state_path("approved"),
+            comment=state_path("approval_comment"),
+        ),
         outcomes=["submitted", "cancelled"],
     )
     skip_email = builder.use(
         mark_email_skipped_spec,
         id="skip_email",
-        out_map={"email_status": state_path("email_status")},
+        out_map=bind_state(email_status=state_path("email_status")),
         desc="Record that email delivery was skipped",
     )
 
@@ -484,3 +484,22 @@ def test_node_decorator_infers_nodereturn_output_model() -> None:
 
 def test_node_decorator_detects_async_automatically() -> None:
     assert inferred_async_echo.is_async is True
+
+
+def test_condition_dsl_compiles_to_core_condition() -> None:
+    condition = state("should_email").eq(True) & exists(state("summary"))
+
+    assert condition.to_condition().model_dump() == {
+        "op": "and",
+        "args": [
+            {
+                "op": "eq",
+                "left": {"path": "state.should_email"},
+                "right": {"value": True},
+            },
+            {
+                "op": "exists",
+                "path": "state.summary",
+            },
+        ],
+    }

wf_authoring/__init__.py

@@ -1,5 +1,7 @@
 from .builder import WorkflowBuilder
 from .catalog import NodeCatalog, NodeCatalogEntry
+from .conditions import context, exists, input, state
+from .mapping import bind_fields, bind_state, merge_maps
 from .paths import GraphPath, context_path, graph_path, input_path, state_path
 from .spec import NodeReturn, NodeSpec, build_registry, node
 
@@ -10,10 +12,17 @@ __all__ = [
     "NodeReturn",
     "NodeSpec",
     "WorkflowBuilder",
+    "bind_fields",
     "build_registry",
+    "bind_state",
+    "merge_maps",
+    "context",
     "context_path",
+    "exists",
     "graph_path",
+    "input",
     "input_path",
     "node",
+    "state",
     "state_path",
 ]

wf_authoring/builder.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from collections.abc import Mapping
 from dataclasses import dataclass, field
 from typing import Any, Literal, TypeAlias
 
@@ -13,27 +14,32 @@ from wf_core import (
     StateSchema,
     Workflow,
 )
+from wf_core.model import Condition as CoreCondition
 
+from .conditions import Expr, compile_condition
+from .mapping import PathArg
 from .paths import GraphPath
 from .spec import NodeSpec
 
-PathArg: TypeAlias = str | GraphPath
 StepRef: TypeAlias = str | NodeUse | ConditionNode | ForeachNode | InterruptNode
+MapArg: TypeAlias = Mapping[Any, Any]
 
 
-def _normalize_path(path: PathArg) -> str:
-    if isinstance(path, GraphPath):
-        return path.value
-    return path
+def _coerce_path(value: object) -> str:
+    if isinstance(value, str):
+        return value
+    if isinstance(value, GraphPath):
+        return value.value
+    raise TypeError(f"unsupported graph path value {value!r}")
 
 
 def _normalize_mapping(
-    mapping: dict[PathArg, PathArg] | None,
+    mapping: MapArg | None,
 ) -> dict[str, str]:
     if mapping is None:
         return {}
     return {
-        _normalize_path(source): _normalize_path(destination)
+        _coerce_path(source): _coerce_path(destination)
         for source, destination in mapping.items()
     }
 
@@ -60,8 +66,8 @@ class WorkflowBuilder:
         spec: NodeSpec[Any, Any],
         *,
         id: str,
-        in_map: dict[PathArg, PathArg] | None = None,
-        out_map: dict[PathArg, PathArg] | None = None,
+        in_map: MapArg | None = None,
+        out_map: MapArg | None = None,
         desc: str | None = None,
     ) -> NodeUse:
         self.node_specs[spec.name] = spec
@@ -76,8 +82,12 @@ class WorkflowBuilder:
         self.nodes.append(node)
         return node
 
-    def condition(self, *, id: str, check: Any) -> ConditionNode:
-        node = ConditionNode(id=id, type="condition", check=check)
+    def condition(self, *, id: str, check: CoreCondition | Expr) -> ConditionNode:
+        node = ConditionNode(
+            id=id,
+            type="condition",
+            check=compile_condition(check),
+        )
         self.nodes.append(node)
         return node
 
@@ -94,7 +104,7 @@ class WorkflowBuilder:
             {
                 "id": id,
                 "type": "foreach",
-                "over": _normalize_path(over),
+                "over": _coerce_path(over),
                 "as": as_,
                 "mode": mode,
                 "on_item_error": on_item_error,
@@ -108,8 +118,8 @@ class WorkflowBuilder:
         *,
         id: str,
         kind: str,
-        request_map: dict[PathArg, PathArg] | None = None,
-        out_map: dict[PathArg, PathArg] | None = None,
+        request_map: MapArg | None = None,
+        out_map: MapArg | None = None,
         outcomes: list[str] | None = None,
     ) -> InterruptNode:
         node = InterruptNode(

wf_authoring/conditions.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Literal
+
+from wf_core.model import (
+    BinaryCondition,
+    Condition,
+    ExistsCondition,
+    LiteralOperand,
+    NotCondition,
+    PathOperand,
+    VariadicCondition,
+)
+
+
+def _operand(value: object) -> PathOperand | LiteralOperand:
+    if isinstance(value, PathExpr):
+        return PathOperand(path=value.path)
+    return LiteralOperand(value=value)
+
+
+@dataclass(frozen=True, slots=True)
+class Expr:
+    condition: Condition
+
+    def __and__(self, other: object) -> Expr:
+        if not isinstance(other, Expr):
+            return NotImplemented
+        return Expr(VariadicCondition(op="and", args=[self.condition, other.condition]))
+
+    def __or__(self, other: object) -> Expr:
+        if not isinstance(other, Expr):
+            return NotImplemented
+        return Expr(VariadicCondition(op="or", args=[self.condition, other.condition]))
+
+    def __invert__(self) -> Expr:
+        return Expr(NotCondition(op="not", arg=self.condition))
+
+    def to_condition(self) -> Condition:
+        return self.condition
+
+
+@dataclass(frozen=True, slots=True)
+class PathExpr:
+    path: str
+
+    def _binary(self, op: Literal["eq", "ne", "gt", "lt"], other: object) -> Expr:
+        return Expr(
+            BinaryCondition(
+                op=op,
+                left=PathOperand(path=self.path),
+                right=_operand(other),
+            )
+        )
+
+    def eq(self, other: object) -> Expr:
+        return self._binary("eq", other)
+
+    def ne(self, other: object) -> Expr:
+        return self._binary("ne", other)
+
+    def gt(self, other: object) -> Expr:
+        return self._binary("gt", other)
+
+    def lt(self, other: object) -> Expr:
+        return self._binary("lt", other)
+
+    def __eq__(self, other: object) -> Expr:  # ty: ignore[invalid-method-override]
+        return self._binary("eq", other)
+
+    def __ne__(self, other: object) -> Expr:  # ty: ignore[invalid-method-override]
+        return self._binary("ne", other)
+
+    def __gt__(self, other: object) -> Expr:
+        return self.gt(other)
+
+    def __lt__(self, other: object) -> Expr:
+        return self.lt(other)
+
+
+def state(field: str) -> PathExpr:
+    return PathExpr(path=f"state.{field}")
+
+
+def input(field: str) -> PathExpr:
+    return PathExpr(path=f"input.{field}")
+
+
+def context(field: str) -> PathExpr:
+    return PathExpr(path=f"context.{field}")
+
+
+def exists(value: PathExpr) -> Expr:
+    return Expr(ExistsCondition(op="exists", path=value.path))
+
+
+def compile_condition(value: Condition | Expr) -> Condition:
+    if isinstance(value, Expr):
+        return value.to_condition()
+    return value

wf_authoring/mapping.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import TypeAlias
+
+from .paths import GraphPath
+
+PathArg: TypeAlias = str | GraphPath
+
+
+def normalize_path(path: PathArg) -> str:
+    if isinstance(path, GraphPath):
+        return path.value
+    return path
+
+
+def bind_fields(**mapping: PathArg) -> dict[str, str]:
+    return {
+        normalize_path(source): destination
+        for destination, source in mapping.items()
+    }
+
+
+def bind_state(**mapping: PathArg) -> dict[str, str]:
+    return {
+        destination: normalize_path(target)
+        for destination, target in mapping.items()
+    }
+
+
+def merge_maps(*maps: Mapping[str, str]) -> dict[str, str]:
+    merged: dict[str, str] = {}
+    for mapping in maps:
+        merged.update(mapping)
+    return merged