Skip to content

01 – Basic Linear Pipeline

A fixed sequence of steps: each step receives every state the previous step produced, and the last step's states are the terminal results.

🎯 Goal

Normalize and sum a list of numbers with a three-node chain: Clean → Sum → Announce.

🧱 Nodes

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from dagpipe import Node, State

class Clean(Node):
    id = "math.clean"

    def resolve(self, state: State):
        numbers = [n for n in state.get("numbers", []) if isinstance(n, (int, float))]
        yield self.fork(state, payload_update={"numbers": numbers})

class Sum(Node):
    id = "math.sum"

    def resolve(self, state: State):
        yield self.fork(
            state,
            payload_update={"total": sum(state.get("numbers", []))},
        )

class Announce(Node):
    id = "math.announce"

    def resolve(self, state: State):
        print(f"total={state.get('total')}")
        yield self.fork(state, payload_update={"announced": True})

🏃 Run it

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
from dagpipe import Engine, Payload, Schema, State

class CalcState(State):
    schema = Schema({
        "numbers": list,
        "total": int | float | None,
        "announced": bool | None,
    })

engine = Engine([Clean(), Sum(), Announce()])
results = engine.run(CalcState(payload=Payload({"numbers": [1, "x", 2, 3.5]})))

state = results[0]
assert state.get("numbers") == [1, 2, 3.5]
assert state.get("total") == 6.5
assert state.get("announced") is True
assert len(state.history) == 3      # clean → sum → announce
assert state.depth == 3

🔍 What just happened

  • The engine seeded states = [root].
  • Clean yielded one new state → states = [clean_state].
  • Sum consumed it → states = [sum_state].
  • Announce consumed it → states = [announce_state] (terminal).

If any node yields zero states, the engine breaks and returns an empty list — that is how a pipeline can terminate early.

💡 Tips

  • Order in the sequence is the execution order — keep stateful order in mind.
  • Every node sees all states produced by the previous node. If a node forks twice, the next step runs once per fork.
  • Prefer Node.fork over State.fork inside resolve() so the node ID lands in history automatically.