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:

Name	Type	Description
int	int	Hash of the node ID, allowing nodes to be used as dict keys.

__new__

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

Create or reuse an async node instance.

Parameters:

Name	Type	Description	Default
*args	Any	Positional constructor arguments forwarded to __init__.	()
**kwargs	Any	Keyword constructor arguments forwarded to __init__.	{}

Returns:

Name	Type	Description
AsyncNode	AsyncNode	A fresh instance for subclasses declaring a parameterized __init__, or the shared singleton for stateless subclasses.

__repr__

__repr__() -> str

Return computation identity based on node ID.

Returns:

Name	Type	Description
str	str	String of the form <Node {id}>.

__str__

__str__() -> str

Return user-facing display name.

Returns:

Name	Type	Description
str	str	String of the form <Node {name}>.

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.

Returns:

Name	Type	Description
bool	bool	True if the node is an AsyncNode instance.

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

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

Execute no-op resolution in sync contexts.

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

Parameters:

Name	Type	Description	Default
state	State	Input execution state.	required

Returns:

Type	Description
Iterable[State]	Iterable[State]: Empty tuple, since async execution is handled by resolve_async.

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

Guarantees:

- 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.

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_async() yields a non-State object.

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:

Name	Type	Description
int	int	Hash of the node ID, allowing nodes to be used as dict keys.

__new__

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

Create or reuse a node instance.

Parameters:

Name	Type	Description	Default
*args	Any	Positional constructor arguments forwarded to __init__.	()
**kwargs	Any	Keyword constructor arguments forwarded to __init__.	{}

Returns:

Name	Type	Description
Node	Node	A fresh instance for subclasses declaring a parameterized __init__, or the shared singleton for stateless subclasses.

Notes

Guarantees:

- 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 computation identity based on node ID.

Returns:

Name	Type	Description
str	str	String of the form <Node {id}>.

__str__

__str__() -> str

Return user-facing display name.

Returns:

Name	Type	Description
str	str	String of the form <Node {name}>.

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.

Returns:

Name	Type	Description
bool	bool	True if the node is an AsyncNode instance.

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.