Skip to content

⛓️ dagpipe — Directed Acyclic Graph Execution for Deterministic State

dagpipe executes pipelines composed of nodes connected in a directed acyclic graph (DAG). Each node receives an immutable State and optionally produces derived states for downstream nodes. Execution is deterministic: the same input state and pipeline always produce the same output states and lineage.

Doc model: this wiki is written for humans — how‑to guides, examples, and testing recipes. The authoritative API contracts live in the code (GSDFC docstrings) and the machine‑readable bundle under docs/mcp/.

🚀 Key Features

  • 🧱 Node-based execution — each node is a pure, deterministic unit of work
  • 🔀 DAG topology — branching, merging, and multiple roots out of the box
  • 🧊 Immutable State — states are never mutated; fork() creates children
  • 🗂️ Hierarchical Payload — dot-path read/update over nested data
  • 📐 Schema validation — payloads and updates validated at construction
  • ⚡ Async execution — AsyncNode + run_async for I/O-bound steps
  • 🪜 Step-wise runs — resumable, progress-hooked execution via run_steps
  • 📜 YAML pipelines — declare schema, nodes, and graph in one file (load_pipeline)

📦 Installation

From your internal PyPI:

1
pip install --extra-index-url https://$PYPI_USERNAME:$PYPI_PASSWORD@pip.aetoskia.com/simple dagpipe

From local source:

1
pip install -e .

⚡ Quick Start

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

class HelloNode(Node):
    id = "hello"

    def resolve(self, state):
        yield self.fork(state, payload_update={"msg": "hello"})

class MyState(State):
    schema = Schema({"msg": str | None})

graph = Graph()
graph.add_root(HelloNode())

engine = Engine(graph)
results = engine.run(MyState(payload=Payload({})))

assert results[0].get("msg") == "hello"

📁 Documentation Structure

Section Description
Overview Architecture, lifecycle, and execution model decision table
Core Components Validated reference for the public API surface
Use cases Step-by-step recipes, grouped by theme
· Getting Started
· 01 – Basic linear pipeline Sequential multi-node pipeline
· 02 – Graph execution Branching, merging, multiple roots
· 03 – YAML pipelines Declarative pipelines via load_pipeline
· State & Data
· 04 – Schema and payload Nested schemas and dot-path access
· 05 – State fork and lineage Immutability, lineage, confidence
· Async & Steps
· 06 – Async execution AsyncNode and run_async
· 07 – Step-wise execution Resumable runs and progress hooks
Best Practices Node design, state design, and gotchas
Error Handling Library exceptions and where they surface
Testing Mock-based quickstart

© Aetoskia Internal — dagpipe 0.0.1