more std, and buffing @node

This commit is contained in:
lda
2026-05-06 23:17:45 +07:00 Verified
parent 1f8a22d2a9
commit 8d3b88c6e7
7 changed files with 288 additions and 7 deletions
+19 -1
View File
@@ -12,17 +12,33 @@ from .sequences import (
last_item_or_none,
length,
)
from .values import CoalesceInput, ValueOutput, coalesce
from .values import (
CoalesceInput,
ConstantInput,
PickKeyInput,
TruthyInput,
ValueOutput,
coalesce,
constant,
default_if_none,
pick_key,
truthy,
)
__all__ = [
"BoolOutput",
"CoalesceInput",
"ConstantInput",
"CountOutput",
"ItemOutput",
"MaybeItemOutput",
"PickKeyInput",
"SequenceInput",
"TruthyInput",
"ValueOutput",
"coalesce",
"constant",
"default_if_none",
"first_item",
"first_item_maybe",
"first_item_or_none",
@@ -30,4 +46,6 @@ __all__ = [
"last_item",
"last_item_or_none",
"length",
"pick_key",
"truthy",
]
+17
View File
@@ -8,22 +8,32 @@ from wf_authoring.nodes import NodeReturn, node
class SequenceInput(BaseModel):
"""Input model for ops that consume an ordered sequence."""
items: list[Any]
class ItemOutput(BaseModel):
"""Output model for ops that return a selected item."""
item: Any
class MaybeItemOutput(BaseModel):
"""Output model for ops that may not find an item."""
item: Any | None = None
class CountOutput(BaseModel):
"""Output model for ops that return a count."""
count: int
class BoolOutput(BaseModel):
"""Output model for ops that return a boolean value."""
value: bool
@@ -34,6 +44,7 @@ class BoolOutput(BaseModel):
description="Select the first item from a non-empty sequence.",
)
def first_item(input: SequenceInput) -> ItemOutput:
"""Select the first item from a non-empty sequence."""
if not input.items:
raise ValueError("first_item requires at least one item")
return ItemOutput(item=input.items[0])
@@ -46,6 +57,7 @@ def first_item(input: SequenceInput) -> ItemOutput:
description="Select the first item from a sequence, or None when it is empty.",
)
def first_item_or_none(input: SequenceInput) -> ItemOutput:
"""Select the first item from a sequence, or None if it is empty."""
return ItemOutput(item=input.items[0] if input.items else None)
@@ -57,6 +69,7 @@ def first_item_or_none(input: SequenceInput) -> ItemOutput:
description="Select the first item from a sequence, routing to found or missing.",
)
def first_item_maybe(input: SequenceInput) -> NodeReturn[MaybeItemOutput]:
"""Select the first item and route by whether one exists."""
if not input.items:
return NodeReturn(outcome="missing", output=MaybeItemOutput())
return NodeReturn(outcome="found", output=MaybeItemOutput(item=input.items[0]))
@@ -69,6 +82,7 @@ def first_item_maybe(input: SequenceInput) -> NodeReturn[MaybeItemOutput]:
description="Select the last item from a non-empty sequence.",
)
def last_item(input: SequenceInput) -> ItemOutput:
"""Select the last item from a non-empty sequence."""
if not input.items:
raise ValueError("last_item requires at least one item")
return ItemOutput(item=input.items[-1])
@@ -81,6 +95,7 @@ def last_item(input: SequenceInput) -> ItemOutput:
description="Select the last item from a sequence, or None when it is empty.",
)
def last_item_or_none(input: SequenceInput) -> ItemOutput:
"""Select the last item from a sequence, or None if it is empty."""
return ItemOutput(item=input.items[-1] if input.items else None)
@@ -91,6 +106,7 @@ def last_item_or_none(input: SequenceInput) -> ItemOutput:
description="Count the items in a sequence.",
)
def length(input: SequenceInput) -> CountOutput:
"""Count the number of items in a sequence."""
return CountOutput(count=len(input.items))
@@ -101,4 +117,5 @@ def length(input: SequenceInput) -> CountOutput:
description="Return whether a sequence is empty.",
)
def is_empty(input: SequenceInput) -> BoolOutput:
"""Return whether a sequence has no items."""
return BoolOutput(value=not input.items)
+68 -1
View File
@@ -4,15 +4,38 @@ from typing import Any
from pydantic import BaseModel
from wf_authoring.nodes import node
from wf_authoring.nodes import NodeReturn, node
class CoalesceInput(BaseModel):
"""Input model for selecting the first non-None value."""
value: Any | None = None
fallback: Any
class ValueOutput(BaseModel):
"""Output model for ops that emit an arbitrary value."""
value: Any
class ConstantInput(BaseModel):
"""Input model for passing through a configured value."""
value: Any
class PickKeyInput(BaseModel):
"""Input model for selecting a value from a mapping by key."""
mapping: dict[str, Any]
key: str
class TruthyInput(BaseModel):
"""Input model for routing by Python truthiness."""
value: Any
@@ -23,4 +46,48 @@ class ValueOutput(BaseModel):
description="Return value when it is not None, otherwise return fallback.",
)
def coalesce(input: CoalesceInput) -> ValueOutput:
"""Return value when it is not None, otherwise return fallback."""
return ValueOutput(value=input.value if input.value is not None else input.fallback)
default_if_none = node(
name="authoring.default_if_none",
description="Alias for coalesce: return fallback only when value is None.",
)(coalesce)
"""Alias for coalesce with a more explicit name for None-defaulting workflows."""
@node(
name="authoring.constant",
input_model=ConstantInput,
output_model=ValueOutput,
description="Return the provided value unchanged.",
)
def constant(input: ConstantInput) -> ValueOutput:
"""Return the provided value unchanged."""
return ValueOutput(value=input.value)
@node(
name="authoring.pick_key",
input_model=PickKeyInput,
output_model=ValueOutput,
description="Select a value from a mapping by key, returning None if missing.",
)
def pick_key(input: PickKeyInput) -> ValueOutput:
"""Select a value from a mapping by key, returning None if missing."""
return ValueOutput(value=input.mapping.get(input.key))
@node(
name="authoring.truthy",
input_model=TruthyInput,
output_model=ValueOutput,
outcomes=("truthy", "falsey"),
description="Route based on Python truthiness of a value.",
)
def truthy(input: TruthyInput) -> NodeReturn[ValueOutput]:
"""Route based on Python truthiness of a value."""
value = bool(input.value)
outcome = "truthy" if value else "falsey"
return NodeReturn(outcome=outcome, output=ValueOutput(value=value))