Skip to content

🧹 Best Practices

Conventions adopted across dagpipe pipelines. Follow these to keep graphs deterministic, observable, and debuggable.

🧱 Node design

  • Keep nodes pure. A node must never mutate its input State — fork a new one instead. Inputs are shared across branches, so mutation would corrupt siblings.
  • Yield zero states to prune. return () in resolve kills the branch: children never run and this path contributes no terminal state.
  • One responsibility per node. Name nodes for what they do (entity.resolve_numeric_merchant, text.normalize), mirroring their file.module location.
  • Prefer stateless subclasses. The singleton-per-subclass model means a stateless node can be reused everywhere for free. Only declare __init__ when you need per-run dependencies (then inspect how instances are created — custom __init__ opts out of the singleton).
  • Use Node.fork inside resolve so the node ID is recorded in history automatically.

🎯 State & schema design

  • Validate as early as possible. A root State validates its payload in __post_init__ — construct roots inside a factory with a clear error path.
  • Declare the schema once at the Schema-building layer and reuse it across state subclasses and YAML definitions.
  • Use object sparingly. Free-form object fields defeat validation. Prefer explicit types or unions (str | None) wherever the contract is known.
  • Evolve schemas deliberately. fork() updates are path-validated; adding a new path is a breaking change for existing payloads.

🔀 Graph patterns

  • Multiple roots are independent. They receive the same root state and share nothing — don't expect cross-communication between roots.
  • Merge = shared child. To join branches, point all of them at the same node. The merge node runs once per incoming state.
  • Prune early with guards. if not state.get(...): return in resolve keeps downstream work minimal.
  • Keep graphs acyclic by construction. Cycle detection exists, but a cycle is always a design bug. Lay out your graph like a DAG from the start.
  • Prefer programmatic Graph for dynamic topologies; use YAML only for static pipelines you can diff in review.

⚡ Async guidance

  • Mix sync and async nodes freely — dispatch is per node, not per engine.
  • Wrap per-node I/O in AsyncNode.resolve_async; keep the engine generic.
  • Don't await run_async(...) for all-sync graphs; run is cheaper.

🪜 Step-wise guidance

  • resume_from is 0-based — persist last_index + 1 and resume from it.
  • Treat completed=False as "no output", not "failure".
  • Route ProgressMessage through on_step hooks for tracing/progress bars without coupling the engine.

🧪 Testability

  • Build synthetic State via a small factory (see the make_state fixture pattern in Testing) — one call per scenario.
  • Assert on terminal states and their history, not intermediate prints.
  • For YAML pipelines, test with a tiny importable node module and a tmp_path fixture (as the integration suite does), keeping tests offline.

❌ Anti-patterns

Pattern Why it's wrong
Mutating state.payload or state.metadata Breaks immutability; shared across branches
Building nodes with heavy __init__ Skips singleton reuse; couples pipeline to instance state
Logging inside resolve Hard to test; pollutes output. Use on_step hooks instead
Reusing one Graph across concurrent runs Assumes nodes are stateless — they must be