tidydag 0.1.9

Lightweight, asynchronous Directed Acyclic Graph (DAG) orchestrator for Python.

TidyDAG

Lightweight, asynchronous Directed Acyclic Graph (DAG) orchestrator for Python.

TidyDAG allows you to define complex task dependencies and execute them concurrently while maintaining a shared state. Built on asyncio, it provides a simple API to resolve execution order automatically via topological sorting.

---

📦 Features

• ⚡ Asynchronous Execution: Built on asyncio for non-blocking task orchestration.
• 🔗 Topological Sorting: Automatically resolves execution order based on dependencies.
• 🧠 Shared Context: Pass and mutate state across your graph nodes seamlessly.
• 🛡️ Type Safety: Built with modern Python 3.13+ features and comprehensive type hints.
• 🧩 Simple API: Define nodes and dependencies with minimal boilerplate.
• 🤖 Task Automation: Managed with uv, ruff, and Taskfile.yml.

---

📁 Project Layout

.
├── src/                          # Core library source code
│   └── tidydag/
│       ├── node/                 # Base Node and state definitions
│       └── orchestrator/         # DAG execution engine
├── test/                         # Pytest-based test suite
│   └── unit/
├── Taskfile.yml                  # Task automation
├── pyproject.toml                # Python project config
└── README.md

---

🔧 Setup

1. Install dependencies (requires Python ≥ 3.13 and uv):

   task deps:install
   

---

💡 Core Concepts

1. Node

The basic building block of a DAG. Every task must inherit from the Node class and implement the execute method.

2. Orchestrator

The engine that manages the execution of the graph. It ensures nodes are executed only after their parents have successfully completed.

3. OrchestratorContext

A container for shared state that is passed to every node during execution.

---

🚀 Quick Start

Defining Custom Nodes

from dataclasses import dataclass
from tidydag.node.base import Node, NodeState, SuccessState, OrchestratorContext

class HelloNode(Node):
    async def execute(self, ctx: OrchestratorContext) -> NodeState:
        print(f"Hello from {self.name}!")
        return SuccessState()

class ProcessNode(Node):
    async def execute(self, ctx: OrchestratorContext) -> NodeState:
        ctx.state.count += 1
        return SuccessState()

Running the DAG

from tidydag.orchestrator import Orchestrator, OrchestratorContext

@dataclass
class HelloContext(OrchestratorContext):
    count: int = 0

# Initialize the orchestrator
orchestrator = Orchestrator()

# Define nodes and dependencies
node_a = HelloNode(name="A")
node_b = ProcessNode(name="B", parents=node_a)
node_c = HelloNode(name="C", parents=node_a)
node_d = HelloNode(name="D", parents=[node_b, node_c])

# Add nodes to the orchestrator
for node in [node_a, node_b, node_c, node_d]:
    orchestrator.add_node(node)

initial_state = HelloContext()
# Run the graph with an initial state
orchestrator.run_sync(state=initial_state)

print(f"Final state: {initial_state}")

---

🧪 Testing

Run all tests with coverage:

task test:run

To generate a coverage badge (docs/coverage.svg), run:

task test:run BADGE=true

---

✨ Automation (Taskfile)

Common commands available via task:

task                    # Show available tasks
task deps:install       # Create venv and sync Python dependencies
task test:run           # Run all tests with pytest
task lint:check         # Run linters and formatters in check mode
task lint:format        # Format code with ruff

---

📄 License

Proprietary — ©Taidy. All rights reserved.

This project is intended for internal use only and may not be copied, distributed, or modified outside of authorized teams without written permission from Taidy.