Node

dagpipe.node

Summary

Defines the Node abstraction used by dagpipe.

A node represents a single unit of pipeline execution logic. It consumes one State and produces zero, one, or many new State objects.

Nodes are connected using a Graph and executed by an Engine.

---

Design principles

• Pure: Must not mutate input state.
• Deterministic: Same input produces same output.
• Stateless: Recommended to be stateless for reuse.
• Composable: Nodes enable branching execution graphs.

Classes

AsyncNode

Bases: Node

Base class for nodes whose execution is asynchronous.

Subclasses implement resolve_async (an async generator yielding derived State objects). The engine dispatches to resolve_async when running an async traversal (see Engine.run_async).

Sync-only engines (and the base Node.run) treat an AsyncNode as a no-op consumer: calling run on an AsyncNode returns no states, signalling that an async engine is required.

Functions

__hash__

__hash__() -> int

Return stable hash based on node ID.

Returns

int

__repr__

__repr__() -> str

Return debug representation.

Returns

str

__str__

__str__() -> str

Return display representation.

Returns

str

clean_id_and_name classmethod

clean_id_and_name() -> None

Normalize and validate node ID and display name.

Raises:

Type	Description
TypeError	If ID is not a string.
ValueError	If ID format is invalid.

Notes

Guarantees:

- Generates ID from module and class name if missing.
- Validates ID format.
- Generates human-readable name if missing.

fork

fork(
    state: State,
    *,
    payload_update: Any = None,
    confidence_delta: float = 0.0,
    metadata_update: Any = None
) -> State

Create a child State attributed to this node.

Parameters:

Name	Type	Description	Default
state	State	Parent execution state.	required
payload_update	Any	Dot-path payload updates.	None
confidence_delta	float	Confidence adjustment.	0.0
metadata_update	Any	Metadata updates.	None

Returns:

Name	Type	Description
State	State	New child execution state.

Notes

Responsibilities:

- Convenience wrapper around `State.fork()` that automatically
  records this node's ID in state history.

is_async

is_async() -> bool

Return whether this node executes asynchronously.

node_id_to_name staticmethod

node_id_to_name(node_id: str) -> str

Convert a dotted snake_case node ID into a human-readable name.

Parameters:

Name	Type	Description	Default
node_id	str	Unique node identifier (e.g., 'entity.resolve.numeric_merchant').	required

Returns:

Name	Type	Description
str	str	Human-readable display name (e.g., 'Entity › Resolve › Numeric Merchant').

resolve_async async

resolve_async(state: State) -> Iterable[State]

Execute node logic asynchronously.

Parameters:

Name	Type	Description	Default
state	State	Input execution state.	required

Returns:

Type	Description
Iterable[State]	Iterable[State]: Derived execution state(s).

Notes

Subclasses implement this. Must not mutate the input state. Should use fork() to create child states.

run

run(state: State) -> tuple[State, ...]

Execute this node on a State.

Parameters:

Name	Type	Description	Default
state	State	Input execution state.	required

Returns:

Type	Description
tuple[State, ...]	tuple[State, ...]: Derived execution states.

Raises:

Type	Description
TypeError	If resolve() yields a non-State object.

run_async async

run_async(state: State) -> tuple[State, ...]

Execute this node asynchronously on a state, validating outputs.

Node

Bases: ABC

Base class for all dagpipe execution nodes.

Attributes:

Name	Type	Description
id	str	Unique identifier of the node (snake_case dotted format).
name	str	Human-readable display name.

Notes

Responsibilities:

- Represents a deterministic unit of execution in the pipeline graph.
- Consumes one `State` and produces zero, one, or many derived states.
- Defines execution logic and enables branching, filtering, and transformation.

Guarantees:

- Nodes must never mutate the input `State`.
- Instances are singletons per subclass and reused across executions.

Functions

__hash__

__hash__() -> int

Return stable hash based on node ID.

Returns

int

__new__

__new__(*args: Any, **kwargs: Any) -> Node

Create or reuse a Node instance.

Stateless subclasses (no parameterized __init__) share one singleton instance per class — matching the original dagpipe behaviour underpinning set_registry-style configuration. Subclasses that declare an __init__ requiring instance-state arguments get a fresh instance per construction so pipeline builders can inject per-run dependencies.

__repr__

__repr__() -> str

Return debug representation.

Returns

str

__str__

__str__() -> str

Return display representation.

Returns

str

clean_id_and_name classmethod

clean_id_and_name() -> None

Normalize and validate node ID and display name.

Raises:

Type	Description
TypeError	If ID is not a string.
ValueError	If ID format is invalid.

Notes

Guarantees:

- Generates ID from module and class name if missing.
- Validates ID format.
- Generates human-readable name if missing.

fork

fork(
    state: State,
    *,
    payload_update: Any = None,
    confidence_delta: float = 0.0,
    metadata_update: Any = None
) -> State

Create a child State attributed to this node.

Parameters:

Name	Type	Description	Default
state	State	Parent execution state.	required
payload_update	Any	Dot-path payload updates.	None
confidence_delta	float	Confidence adjustment.	0.0
metadata_update	Any	Metadata updates.	None

Returns:

Name	Type	Description
State	State	New child execution state.

Notes

Responsibilities:

- Convenience wrapper around `State.fork()` that automatically
  records this node's ID in state history.

is_async

is_async() -> bool

Return whether this node executes asynchronously.

node_id_to_name staticmethod

node_id_to_name(node_id: str) -> str

Convert a dotted snake_case node ID into a human-readable name.

Parameters:

Name	Type	Description	Default
node_id	str	Unique node identifier (e.g., 'entity.resolve.numeric_merchant').	required

Returns:

Name	Type	Description
str	str	Human-readable display name (e.g., 'Entity › Resolve › Numeric Merchant').

resolve abstractmethod

resolve(state: State) -> Iterable[State]

Execute node logic.

Parameters:

Name	Type	Description	Default
state	State	Input execution state.	required

Returns:

Type	Description
Iterable[State]	Iterable[State]: Derived execution state(s).

Notes

Responsibilities:

- Subclasses implement specific resolution behavior.
- Must not mutate input state.
- Should use `fork()` to create child states.
- May yield zero states to terminate a branch.

run

run(state: State) -> tuple[State, ...]

Execute this node on a State.

Parameters:

Name	Type	Description	Default
state	State	Input execution state.	required

Returns:

Type	Description
tuple[State, ...]	tuple[State, ...]: Derived execution states.

Raises:

Type	Description
TypeError	If resolve() yields a non-State object.