Skip to content

03 – YAML Pipelines

load_pipeline() builds a complete pipeline — schema, state subclass, nodes, graph, engine, and initial payload — from one declarative YAML file.

🎯 Goal

Recreate the entity-resolution DAG from use case 02 purely in YAML.

📄 The YAML file

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
version: 1

schema:
  raw: object
  tags: list | None
  entity:
    candidate: str | None
    name: str | None

initial:
  raw: {}

nodes:
  extract:
    class: testnodes.ExtractCandidate
  resolve_amzn:
    class: testnodes.ResolveAmazon
  resolve_fk:
    class: testnodes.ResolveFlipkart
  tag:
    class: testnodes.TagNode
  direct:
    class: testnodes.DirectNode
  kill:
    class: testnodes.KillNode

graph:
  roots:
    - extract
    - direct
    - kill
  edges:
    extract:
      - resolve_amzn
      - resolve_fk
    resolve_amzn:
      - tag
    resolve_fk:
      - tag

Node class paths are fully qualified (module.ClassName). The modules are imported with importlib at load time, so they must be importable from your runtime environment.

🏃 Run it

1
2
3
4
5
6
7
from dagpipe import load_pipeline

pipeline = load_pipeline("pipeline.yaml")
results = pipeline.run()

names = sorted(s.get("entity.name") for s in results)
assert names == ["Amazon", "Direct", "Flipkart"]

🔍 Schema short-hand

Leaf values in schema: map to Python types:

YAML Python
str str
int / float int / float
bool bool
dict / list dict / list
bytes bytes
object object
None NoneType
str | None str \| None (PEP-604 union)

Nested schema blocks become nested Schema instances.

🔍 Overriding the initial payload

Pipeline.run(payload_override=None) merges override values on top of initial: using Payload.update — the root payload is the override, not the default:

1
results = pipeline.run(payload_override={"raw": "custom"})

💡 Tips

  • Give node names that match their class intent — they become the dict keys used by roots/edges.
  • Every node listed in graph.roots/graph.edges must exist in nodes: or you get a KeyError.
  • YAML pipelines are best for static topologies. For dynamic graphs (nodes built at import time, conditional wiring) use the programmatic API.