path handling primitives

tests/test_demo_workflow.py

@@ -13,7 +13,15 @@ from wf_core import (
 )
 from wf_core.demo_workflow import build_demo_registry, build_demo_workflow
 from wf_core.run_factory import create_run_state
-from wf_authoring import NodeReturn, WorkflowBuilder, build_registry, node
+from wf_authoring import (
+    NodeReturn,
+    WorkflowBuilder,
+    build_registry,
+    context_path,
+    input_path,
+    node,
+    state_path,
+)
 
 
 class DriveListFilesInput(BaseModel):
@@ -159,35 +167,35 @@ def build_authoring_demo_workflow():
         start="list_files",
     )
 
-    builder.use(
+    list_files = builder.use(
         drive_list_files_spec,
         id="list_files",
-        in_map={"input.folder_id": "folder_id"},
+        in_map={input_path("folder_id"): "folder_id"},
-        out_map={"documents": "state.documents"},
+        out_map={"documents": state_path("documents")},
         desc="List files from a Google Drive folder",
     )
-    builder.foreach(
+    summarize_each = builder.foreach(
         id="summarize_each",
-        over="state.documents",
+        over=state_path("documents"),
         as_="document",
         mode="serial",
         on_item_error="fail",
     )
-    builder.use(
+    summarize_one = builder.use(
         summarize_document_spec,
         id="summarize_one",
-        in_map={"context.document": "document"},
+        in_map={context_path("document"): "document"},
-        out_map={"item_summary": "state.item_summaries"},
+        out_map={"item_summary": state_path("item_summaries")},
         desc="Summarize one document",
     )
-    builder.use(
+    combine_summaries = builder.use(
         combine_summaries_spec,
         id="combine_summaries",
-        in_map={"state.item_summaries": "item_summaries"},
+        in_map={state_path("item_summaries"): "item_summaries"},
-        out_map={"summary": "state.summary"},
+        out_map={"summary": state_path("summary")},
         desc="Combine item summaries into one final summary",
     )
-    builder.condition(
+    should_email = builder.condition(
         id="should_email",
         check={
             "op": "eq",
@@ -195,44 +203,44 @@ def build_authoring_demo_workflow():
             "right": {"value": True},
         },
     )
-    builder.use(
+    send_email = builder.use(
         send_email_spec,
         id="send_email",
-        in_map={"state.summary": "summary"},
+        in_map={state_path("summary"): "summary"},
-        out_map={"email_status": "state.email_status"},
+        out_map={"email_status": state_path("email_status")},
         desc="Send the summary by email",
     )
-    builder.interrupt(
+    approve_email = builder.interrupt(
         id="approve_email",
         kind="approval",
         request_map={
-            "state.summary": "summary",
+            state_path("summary"): "summary",
-            "input.folder_id": "folder_id",
+            input_path("folder_id"): "folder_id",
         },
         out_map={
-            "approved": "state.approved",
+            "approved": state_path("approved"),
-            "comment": "state.approval_comment",
+            "comment": state_path("approval_comment"),
         },
         outcomes=["submitted", "cancelled"],
     )
-    builder.use(
+    skip_email = builder.use(
         mark_email_skipped_spec,
         id="skip_email",
-        out_map={"email_status": "state.email_status"},
+        out_map={"email_status": state_path("email_status")},
         desc="Record that email delivery was skipped",
     )
 
-    builder.connect("list_files", "ok", "summarize_each")
+    builder.connect(list_files, "ok", summarize_each)
-    builder.connect("summarize_each", "loop", "summarize_one")
+    builder.connect(summarize_each, "loop", summarize_one)
-    builder.connect("summarize_each", "done", "combine_summaries")
+    builder.connect(summarize_each, "done", combine_summaries)
-    builder.connect("summarize_one", "ok", END)
+    builder.connect(summarize_one, "ok", END)
-    builder.connect("combine_summaries", "ok", "should_email")
+    builder.connect(combine_summaries, "ok", should_email)
-    builder.connect("should_email", "true", "approve_email")
+    builder.connect(should_email, "true", approve_email)
-    builder.connect("should_email", "false", "skip_email")
+    builder.connect(should_email, "false", skip_email)
-    builder.connect("approve_email", "submitted", "send_email")
+    builder.connect(approve_email, "submitted", send_email)
-    builder.connect("approve_email", "cancelled", "skip_email")
+    builder.connect(approve_email, "cancelled", skip_email)
-    builder.connect("send_email", "sent", END)
+    builder.connect(send_email, "sent", END)
-    builder.connect("skip_email", "ok", END)
+    builder.connect(skip_email, "ok", END)
 
     registry = build_registry(
         drive_list_files_spec,

wf_authoring/__init__.py

@@ -1,13 +1,19 @@
 from .builder import WorkflowBuilder
 from .catalog import NodeCatalog, NodeCatalogEntry
+from .paths import GraphPath, context_path, graph_path, input_path, state_path
 from .spec import NodeReturn, NodeSpec, build_registry, node
 
 __all__ = [
     "NodeCatalog",
     "NodeCatalogEntry",
+    "GraphPath",
     "NodeReturn",
     "NodeSpec",
     "WorkflowBuilder",
     "build_registry",
+    "context_path",
+    "graph_path",
+    "input_path",
     "node",
+    "state_path",
 ]

wf_authoring/builder.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from typing import Any, Literal
+from typing import Any, Literal, TypeAlias
 
 from wf_core import (
     ConditionNode,
@@ -14,8 +14,35 @@ from wf_core import (
     Workflow,
 )
 
+from .paths import GraphPath
 from .spec import NodeSpec
 
+PathArg: TypeAlias = str | GraphPath
+StepRef: TypeAlias = str | NodeUse | ConditionNode | ForeachNode | InterruptNode
+
+
+def _normalize_path(path: PathArg) -> str:
+    if isinstance(path, GraphPath):
+        return path.value
+    return path
+
+
+def _normalize_mapping(
+    mapping: dict[PathArg, PathArg] | None,
+) -> dict[str, str]:
+    if mapping is None:
+        return {}
+    return {
+        _normalize_path(source): _normalize_path(destination)
+        for source, destination in mapping.items()
+    }
+
+
+def _step_id(ref: StepRef) -> str:
+    if isinstance(ref, str):
+        return ref
+    return ref.id
+
 
 @dataclass(slots=True)
 class WorkflowBuilder:
@@ -33,8 +60,8 @@ class WorkflowBuilder:
         spec: NodeSpec[Any, Any],
         *,
         id: str,
-        in_map: dict[str, str] | None = None,
+        in_map: dict[PathArg, PathArg] | None = None,
-        out_map: dict[str, str] | None = None,
+        out_map: dict[PathArg, PathArg] | None = None,
         desc: str | None = None,
     ) -> NodeUse:
         self.node_specs[spec.name] = spec
@@ -43,8 +70,8 @@ class WorkflowBuilder:
             type="node",
             node=spec.name,
             desc=desc or spec.description,
-            in_map=in_map or {},
+            in_map=_normalize_mapping(in_map),
-            out_map=out_map or {},
+            out_map=_normalize_mapping(out_map),
         )
         self.nodes.append(node)
         return node
@@ -58,7 +85,7 @@ class WorkflowBuilder:
         self,
         *,
         id: str,
-        over: str,
+        over: PathArg,
         as_: str,
         mode: Literal["serial", "parallel"] = "serial",
         on_item_error: Literal["fail", "collect", "skip"] = "fail",
@@ -67,7 +94,7 @@ class WorkflowBuilder:
             {
                 "id": id,
                 "type": "foreach",
-                "over": over,
+                "over": _normalize_path(over),
                 "as": as_,
                 "mode": mode,
                 "on_item_error": on_item_error,
@@ -81,23 +108,27 @@ class WorkflowBuilder:
         *,
         id: str,
         kind: str,
-        request_map: dict[str, str] | None = None,
+        request_map: dict[PathArg, PathArg] | None = None,
-        out_map: dict[str, str] | None = None,
+        out_map: dict[PathArg, PathArg] | None = None,
         outcomes: list[str] | None = None,
     ) -> InterruptNode:
         node = InterruptNode(
             id=id,
             type="interrupt",
             kind=kind,
-            request_map=request_map or {},
+            request_map=_normalize_mapping(request_map),
-            out_map=out_map or {},
+            out_map=_normalize_mapping(out_map),
             outcomes=outcomes or ["submitted"],
         )
         self.nodes.append(node)
         return node
 
-    def connect(self, from_: str, outcome: str, to: str) -> None:
+    def connect(self, from_: StepRef, outcome: str, to: StepRef) -> None:
-        self.edges.append(Edge.model_validate({"from": from_, "outcome": outcome, "to": to}))
+        self.edges.append(
+            Edge.model_validate(
+                {"from": _step_id(from_), "outcome": outcome, "to": _step_id(to)}
+            )
+        )
 
     def compile(self) -> Workflow:
         node_defs = [spec.to_node_def() for spec in self.node_specs.values()]

wf_authoring/paths.py

@@ -0,0 +1,27 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True, slots=True)
+class GraphPath:
+    value: str
+
+    def __str__(self) -> str:
+        return self.value
+
+
+def graph_path(value: str) -> GraphPath:
+    return GraphPath(value)
+
+
+def input_path(field: str) -> GraphPath:
+    return GraphPath(f"input.{field}")
+
+
+def state_path(field: str) -> GraphPath:
+    return GraphPath(f"state.{field}")
+
+
+def context_path(field: str) -> GraphPath:
+    return GraphPath(f"context.{field}")