Skip to content

🧩 Core Components

This page is the validated reference for the public API surface. For step-by-step recipes see the use cases index. For exact signatures and docstrings, see the library reference (docs/lib) or the MCP bundle (docs/mcp).

⚡ Execution Core

Node

Base class for all execution nodes. It is an abstract base class: subclasses implement resolve() and declare an id (dotted snake_case).

1
2
3
4
5
6
7
from dagpipe import Node, State

class Increment(Node):
    id = "math.increment"

    def resolve(self, state: State):
        yield self.fork(state, payload_update={"value": state.get("value") + 1})

Key facts:

  • Singleton per subclass — stateless subclasses share one instance. A() is A() is True. Subclasses that declare a custom __init__ get one instance per construction (use this to inject per-run dependencies).
  • fork() — convenience wrapper around State.fork() that records the node ID into state history.
  • resolve() must yield State — anything else raises TypeError.
  • node_id_to_name(node_id) — converts entity.resolve.numeric_merchant into Entity › Resolve › Numeric Merchant.
  • name — auto-derived from the ID unless set explicitly.

AsyncNode

Base class for asynchronous nodes. Subclasses implement resolve_async() (an async generator yielding State objects). See use case 06.

1
2
3
4
5
6
7
8
from dagpipe import AsyncNode

class FetchRemote(AsyncNode):
    id = "web.fetch"

    async def resolve_async(self, state: State):
        data = await fetch(state.get("url"))
        yield self.fork(state, payload_update={"body": data})

A sync engine that hits an AsyncNode treats it as a no-op (yields no states). Use Engine.run_async when the graph contains AsyncNodes.

Graph

DAG topology container. Stores connectivity only — it never executes nodes.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
from dagpipe import Graph

graph = Graph()
graph.add_root(a)          # node with no parents
graph.add_edge(a, b)       # b is a child of a
graph.add_edge(a, c)
graph.children(a)          # (b, c)
graph.parents(b)           # (a,)
graph.roots()              # nodes with no incoming edges
graph.nodes()              # all registered nodes
  • Cycle detection is automatic: adding an edge that would create a cycle raises ValueError (self-edges too).
  • Nodes are registered implicitly by add_edge / add_root.
  • The graph is mutable during construction but treated as immutable at runtime.

Engine

Orchestrator that runs a linear Sequence[Node] or a Graph.

1
2
3
4
5
from dagpipe import Engine

engine = Engine(graph)                # or Engine([n1, n2, n3])
results = engine.run(root_state)      # list[State] of terminal states
results_async = await engine.run_async(root_state)
  • Linear mode — each node feeds every downstream state it receives, in order.
  • Graph mode — BFS traversal from all roots; states fan out along edges and are collected as terminal states at nodes with no children.
  • Also exposes run_steps / run_steps_async and a nodes property.
  • Never mutates State, Node, or Graph instances.

🧊 State & Data

State

Immutable execution snapshot at one point in traversal. Subclass and bind a schema:

1
2
3
4
from dagpipe import State, Schema

class MyState(State):
    schema = Schema({"value": int, "label": str | None})
  • Validates its payload against schema at construction (SchemaError on violation).
  • fork() is the only supported mechanism for producing a new state. Use State.fork directly, or the Node.fork convenience wrapper.
  • Tracks confidence, parent, depth, and history (ordered node-ID lineage) for observability.
  • lineage() — root-to-this ordered tuple.
  • get(key) / has(key) read dot-paths from the underlying Payload.

Payload

Immutable hierarchical container with dot-path access.

1
2
3
4
5
6
payload = Payload({"user": {"address": {"city": "Mumbai"}}})
payload.get("user.address.city")            # 'Mumbai'
payload.has("user.address.zip")             # False
payload.update({"user.address.zip": 400001})  # new Payload, original untouched
payload.keys()                              # ('user',)
payload.as_dict()                           # read-only view of underlying mapping
  • Updates are atomic and cheap — only modified branches are copied.

Schema

Immutable hierarchical validation tree. Leaf nodes are types or PEP-604 unions; nested Schema instances describe nested structure.

1
2
3
4
5
6
AddressSchema = Schema({"city": str, "zip": int | None})
UserSchema = Schema({"name": str, "address": AddressSchema})

user_payload = Payload({"name": "Ada", "address": {"city": "London"}})
UserSchema.validate_payload(user_payload)   # no-op on success
UserSchema.validate_update({"address.city": "Paris"})
  • validate_payload — full structure check.
  • validate_update — path existence check for fork() updates.

SchemaError

Raised when payload data violates the declared schema: invalid structure, undefined path, or invalid type. See Error Handling.

📜 Declarative Pipelines

Pipeline

Dataclass wrapping engine, state_cls, and initial_payload. Executes with run(payload_override=None) and returns terminal states.

load_pipeline(path)

Factory that builds a Pipeline from one YAML file:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
version: 1
schema:
  value: int
initial:
  value: 1
nodes:
  step1:
    class: mymod.Increment
graph:
  roots:
    - step1

See use case 03 for the full walkthrough.

🔁 Progress Types

  • StepResult — (index, node_id, states, completed) for one executed step.
  • ProgressMessage — keyword-only, optional fields (lines, blocks, count, unit, raw_ocr_line, error, step, status) plus as_dict(). Passed to step hooks for progress reporting.

Both are produced by Engine.run_steps / run_steps_async — see use case 07.