we een have builtin types for you to use

This commit is contained in:
lda
2026-05-06 22:12:55 +07:00 Verified
parent 05a0db63c8
commit f15f5289ef
3 changed files with 171 additions and 1 deletions
+18
View File
@@ -3,12 +3,21 @@ from .catalog import NodeCatalog, NodeCatalogEntry
from .conditions import context, exists, expr, input, state from .conditions import context, exists, expr, input, state
from .mapping import bind_fields, bind_state, merge_maps from .mapping import bind_fields, bind_state, merge_maps
from .ops import ( from .ops import (
BoolOutput,
CoalesceInput,
CountOutput,
ItemOutput, ItemOutput,
MaybeItemOutput, MaybeItemOutput,
SequenceInput, SequenceInput,
ValueOutput,
coalesce,
first_item, first_item,
first_item_maybe, first_item_maybe,
first_item_or_none, first_item_or_none,
is_empty,
last_item,
last_item_or_none,
length,
) )
from .paths import GraphPath, context_path, graph_path, input_path, state_path from .paths import GraphPath, context_path, graph_path, input_path, state_path
from .spec import ( from .spec import (
@@ -25,6 +34,9 @@ from .subgraph import subgraph_node
__all__ = [ __all__ = [
"NodeCatalog", "NodeCatalog",
"NodeCatalogEntry", "NodeCatalogEntry",
"BoolOutput",
"CoalesceInput",
"CountOutput",
"GraphPath", "GraphPath",
"ItemOutput", "ItemOutput",
"MaybeItemOutput", "MaybeItemOutput",
@@ -33,11 +45,13 @@ __all__ = [
"AsyncRegistryHandler", "AsyncRegistryHandler",
"SyncRegistryHandler", "SyncRegistryHandler",
"SequenceInput", "SequenceInput",
"ValueOutput",
"WorkflowBuilder", "WorkflowBuilder",
"bind_fields", "bind_fields",
"build_async_registry", "build_async_registry",
"build_registry", "build_registry",
"bind_state", "bind_state",
"coalesce",
"merge_maps", "merge_maps",
"context", "context",
"context_path", "context_path",
@@ -49,6 +63,10 @@ __all__ = [
"graph_path", "graph_path",
"input", "input",
"input_path", "input_path",
"is_empty",
"last_item",
"last_item_or_none",
"length",
"node", "node",
"state", "state",
"state_path", "state_path",
+69
View File
@@ -19,6 +19,23 @@ class MaybeItemOutput(BaseModel):
item: Any | None = None item: Any | None = None
class CountOutput(BaseModel):
count: int
class BoolOutput(BaseModel):
value: bool
class CoalesceInput(BaseModel):
value: Any | None = None
fallback: Any
class ValueOutput(BaseModel):
value: Any
@node( @node(
name="authoring.first_item", name="authoring.first_item",
input_model=SequenceInput, input_model=SequenceInput,
@@ -54,3 +71,55 @@ def first_item_maybe(input: SequenceInput) -> NodeReturn[MaybeItemOutput]:
if not input.items: if not input.items:
return NodeReturn(outcome="missing", output=MaybeItemOutput()) return NodeReturn(outcome="missing", output=MaybeItemOutput())
return NodeReturn(outcome="found", output=MaybeItemOutput(item=input.items[0])) return NodeReturn(outcome="found", output=MaybeItemOutput(item=input.items[0]))
@node(
name="authoring.last_item",
input_model=SequenceInput,
output_model=ItemOutput,
description="Select the last item from a non-empty sequence.",
)
def last_item(input: SequenceInput) -> ItemOutput:
if not input.items:
raise ValueError("last_item requires at least one item")
return ItemOutput(item=input.items[-1])
@node(
name="authoring.last_item_or_none",
input_model=SequenceInput,
output_model=ItemOutput,
description="Select the last item from a sequence, or None when it is empty.",
)
def last_item_or_none(input: SequenceInput) -> ItemOutput:
return ItemOutput(item=input.items[-1] if input.items else None)
@node(
name="authoring.length",
input_model=SequenceInput,
output_model=CountOutput,
description="Count the items in a sequence.",
)
def length(input: SequenceInput) -> CountOutput:
return CountOutput(count=len(input.items))
@node(
name="authoring.is_empty",
input_model=SequenceInput,
output_model=BoolOutput,
description="Return whether a sequence is empty.",
)
def is_empty(input: SequenceInput) -> BoolOutput:
return BoolOutput(value=not input.items)
@node(
name="authoring.coalesce",
input_model=CoalesceInput,
output_model=ValueOutput,
description="Return value when it is not None, otherwise return fallback.",
)
def coalesce(input: CoalesceInput) -> ValueOutput:
return ValueOutput(value=input.value if input.value is not None else input.fallback)
+84 -1
View File
@@ -7,12 +7,24 @@ from wf_authoring import (
bind_fields, bind_fields,
bind_state, bind_state,
build_registry, build_registry,
coalesce,
first_item, first_item,
first_item_maybe, first_item_maybe,
first_item_or_none, first_item_or_none,
is_empty,
last_item,
last_item_or_none,
length,
state_path, state_path,
) )
from wf_core import RunStatus, SchemaRef, StateField, StateSchema, execute_workflow from wf_core import (
RunStatus,
RuntimeContext,
SchemaRef,
StateField,
StateSchema,
execute_workflow,
)
def _build_first_workflow(use_safe_first: bool = False): def _build_first_workflow(use_safe_first: bool = False):
@@ -114,3 +126,74 @@ def test_first_item_maybe_routes_missing_outcome() -> None:
assert run.status == RunStatus.COMPLETED assert run.status == RunStatus.COMPLETED
assert run.state["item"] is None assert run.state["item"] is None
assert run.trace[0].outcome == "missing" assert run.trace[0].outcome == "missing"
def test_last_item_selects_last_value() -> None:
registry = build_registry(last_item)
result = registry["authoring.last_item"](
{"items": ["a", "b"]},
RuntimeContext(current_node_id="last"),
)
assert result == {"outcome": "ok", "output": {"item": "b"}}
def test_last_item_fails_on_empty_sequence() -> None:
registry = build_registry(last_item)
with pytest.raises(ValueError, match="last_item requires at least one item"):
registry["authoring.last_item"](
{"items": []},
RuntimeContext(current_node_id="last"),
)
def test_last_item_or_none_returns_none_for_empty_sequence() -> None:
registry = build_registry(last_item_or_none)
result = registry["authoring.last_item_or_none"](
{"items": []},
RuntimeContext(current_node_id="last"),
)
assert result == {"outcome": "ok", "output": {"item": None}}
def test_length_counts_items() -> None:
registry = build_registry(length)
result = registry["authoring.length"](
{"items": ["a", "b", "c"]},
RuntimeContext(current_node_id="length"),
)
assert result == {"outcome": "ok", "output": {"count": 3}}
def test_is_empty_detects_empty_sequence() -> None:
registry = build_registry(is_empty)
result = registry["authoring.is_empty"](
{"items": []},
RuntimeContext(current_node_id="is_empty"),
)
assert result == {"outcome": "ok", "output": {"value": True}}
def test_coalesce_returns_value_or_fallback() -> None:
registry = build_registry(coalesce)
ctx = RuntimeContext(current_node_id="coalesce")
present = registry["authoring.coalesce"](
{"value": "x", "fallback": "fallback"},
ctx,
)
missing = registry["authoring.coalesce"](
{"value": None, "fallback": "fallback"},
ctx,
)
assert present == {"outcome": "ok", "output": {"value": "x"}}
assert missing == {"outcome": "ok", "output": {"value": "fallback"}}