Skip to content

07 – Step-wise Execution

Engine.run_steps runs a pipeline one step at a time, yielding a StepResult per executed node. It supports resume (skip already-done steps) and progress hooks (report started / completed / skipped).

🎯 Goal

Run a three-node pipeline, observe progress, and resume a partial run.

🧱 Setup

 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 Engine, Node, Payload, Schema, State

class StepA(Node):
    id = "qa.step_a"
    def resolve(self, state):
        yield self.fork(state, payload_update={"a": True})

class StepB(Node):
    id = "qa.step_b"
    def resolve(self, state):
        if not state.get("a"):
            return                       # prune branch if precondition unmet
        yield self.fork(state, payload_update={"b": True})

class StepC(Node):
    id = "qa.step_c"
    def resolve(self, state):
        yield self.fork(state, payload_update={"c": True})

class QState(State):
    schema = Schema({"a": bool | None, "b": bool | None, "c": bool | None})

engine = Engine([StepA(), StepB(), StepC()])
root = QState(payload=Payload({}))

🏃 Step by step

1
2
for step in engine.run_steps(root):
    print(f"#{step.index} {step.node_id:10} completed={step.completed} states={len(step.states)}")

Output:

1
2
3
#0 qa.step_a completed=True states=1
#1 qa.step_b completed=True states=1
#2 qa.step_c completed=True states=1

Each StepResult exposes:

Field Meaning
index 0-based step ordinal (resume_from is relative to this)
node_id The node that ran in this step
states States produced by this step
completed True if any state was produced

🔍 Resume after interruption

If you already processed #0 and #1, skip ahead:

1
2
for step in engine.run_steps(root, resume_from=2):
    print(step.node_id)      # only qa.step_c

🔍 Progress hooks

Pass on_step (or set it at engine construction) to be notified per step:

1
2
3
4
5
engine = Engine([StepA(), StepB(), StepC()],
                on_step=lambda step, status, msg: print(step, status))

for step in engine.run_steps(root):
    ...

Output:

1
2
3
4
5
qa.step_a started
qa.step_a completed
qa.step_b started
qa.step_b completed
...

A ProgressMessage can be passed as the third argument for richer progress (lines, blocks, count, unit, error, ...). Use as_dict() to surface it in your logging.

The async variant run_steps_async mirrors this API with on_step: AsyncStepHook — see use case 06.

🔍 Step ordering in graph mode

In graph mode, run_steps derives a deterministic topological order (DFS post-order, roots first), one StepResult per node. A node that yields no states marks that step completed=False and the branch stops there.

💡 Tips

  • Use resume_from for idempotent retries: record the last consumed index (e.g. in a DB), then resume the next run from index + 1.
  • completed=False means "no states produced" (pruned/static node) — treat it as "no work available", not "error".
  • Hooks are a good place to emit telemetry (counters, traces) without coupling the engine to your logging stack.