Build a Workflow Orchestration Engine

Introduction

30 Second Summary

Every complex process has steps that must happen in the right order. You cannot frost a cake before it comes out of the oven, and you cannot bake it before the batter is mixed.

In this project, you will build MiniFlow, a mini workflow orchestration engine in Python. You define data pipelines as YAML files with tasks and dependencies, then MiniFlow resolves the right execution order and runs each task with a color-coded terminal dashboard.

What You'll Build

Run python miniflow.py run workflow.yml in your terminal and watch MiniFlow parse your workflow, resolve the dependency graph, and execute each task with live color-coded status updates.

By the end of this project, you'll have:

• A working orchestration engine where you define workflows in YAML with shell and Python task types, run them with a single command, and watch tasks execute with live status updates. This mirrors the core architecture behind production tools like Kestra and Airflow.
• A DAG dependency resolver that uses topological sorting to determine the correct execution order from dependencies, no matter how you arrange tasks in the file. It catches circular dependencies before any task runs.
• A Rich-powered terminal dashboard with color-coded execution lines, upstream-failure handling that automatically skips downstream tasks, and a final summary table showing every task's state and output.
• Secret Mission: add per-task retry logic with exponential backoff so flaky tasks get multiple chances before the engine gives up.

Are there any prerequisites?

This project assumes basic Python knowledge, including functions, classes, and dictionaries. No prior experience with orchestration tools is required.

Before We Start

Before we dive in, let's lock in what you're about to build and why it matters.

Set Up Your Project

Before your orchestration engine can parse workflow files or display colored status updates, it needs a project structure and two Python libraries.

In this step, you'll create a project folder, set up a Python virtual environment, and install PyYAML for reading YAML files and Rich for terminal formatting.

In this step, get ready to:

• Create your project folder and a Python virtual environment.
• Install PyYAML and Rich.
• Verify both libraries are working.

Create the project directory and install dependencies

• Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.

• Move to your Desktop by running this command:

cd ~/Desktop

• Create your project folder and move into it by running these commands:

mkdir miniflow
cd miniflow

• Confirm you're in the new folder by running:

pwd

You should see a path ending in miniflow.

• Create a Python virtual environment and activate it by running the commands for your operating system:

macOS

python3 -m venv venv
source venv/bin/activate

Windows

python -m venv venv
venv\Scripts\activate

What is a virtual environment?

A virtual environment is an isolated Python setup inside your project folder. Any libraries you install here stay contained to this project and won't interfere with other Python projects on your machine.

You should now see (venv) at the start of your terminal prompt. This confirms the virtual environment is active.

Don't see (venv) in your prompt?

• Make sure you selected the correct tab for your operating system. macOS uses source, while Windows uses a direct path.
• On Windows, if you see a permissions error, try running PowerShell as Administrator.
• help me activate my Python virtual environment

• Install PyYAML and Rich by running:

pip install PyYAML==6.0.3 rich==15.0.0

What does this command do?

• PyYAML==6.0.3 installs the PyYAML library at version 6.0.3. Your engine uses this to read workflow definition files written in YAML.
• rich==15.0.0 installs the Rich library at version 15.0.0. Rich gives you tables, colored text, and panels in the terminal without building a web frontend.

You should see output scrolling through the downloads, ending with a line that says Successfully installed followed by the package names.

Seeing a pip error?

• If you see pip: command not found, make sure your virtual environment is active. You should see (venv) in your prompt.
• If the install hangs or times out, check your internet connection and try running the command again.
• help me fix my pip install error

Verify your installation

Both libraries are installed. Before you test them, take a guess: what do you think running a library as a module with the -m flag will show you?

• Test that Rich is working by running:

python -m rich

You should see Rich's demo output fill your terminal with colored text, tables, and panels. This is Rich showing off everything it can render.

• Check that PyYAML is installed at the correct version by running:

python -c "import yaml; print(yaml.version)"

You should see 6.0.3 printed in your terminal.

Not seeing the expected output?

• If you see No module named rich or ModuleNotFoundError, the install did not succeed. Make sure (venv) is in your prompt, then rerun the pip install command from the previous substep.
• If you see a different version number for PyYAML, rerun pip install PyYAML==6.0.3 to pin the correct version.
• help me verify my PyYAML and Rich installation

• Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type Visual Studio Code and press Enter to open it.
• Click File in the top menu bar.
• Click Open Folder.
• Select the miniflow folder on your Desktop, then click Open.

VS Code opens with your empty miniflow project folder in the sidebar. You'll create all your files here starting in the next step.

Your project folder is ready, your libraries are installed, and VS Code is open. Next up, you'll create your first YAML workflow file and write a parser that turns it into a data structure your engine can work with.

Parse Your First Workflow from YAML

Your project directory is set up and your libraries are installed. Now it's time to give MiniFlow something to actually orchestrate.

Before your engine can run any tasks, it needs a way to read workflow definitions. In this step, you'll create a YAML workflow file that describes a data pipeline and build the parser that turns it into a validated Python data structure.

In this step, get ready to:

• Create a YAML workflow file with three data pipeline tasks.
• Write a parser that validates workflow structure and catches errors.
• Add a CLI with a validate command to test your parser.

Create your workflow file

Orchestration tools like Kestra and Airflow let you define pipelines declaratively. You describe what each task does and the engine figures out how to run them.

• Create a new file called workflow.yml in your miniflow project folder by pasting the following code into it:

id: data_pipeline
description: A sample ETL data pipeline

tasks:
  - id: fetch_data
    type: shell
    command: "echo 'Fetching data from API...'"

  - id: validate_data
    type: python
    script: |
      import json
      data = {"users": 42, "status": "ok"}
      print(f"Validated {data['users']} records")

  - id: transform_data
    type: python
    script: |
      print("Transforming data...")
      records = [{"id": i, "value": i * 10} for i in range(5)]
      print(f"Transformed {len(records)} records")

What does this workflow define?

This file describes a three-step data pipeline. Each task has a unique id and a type that tells the engine how to run it.

• fetch_data is a shell task. It runs a terminal command directly.
• validate_data and transform_data are python tasks. They run inline Python scripts.

• Save workflow.yml.
• Confirm the file exists by running:

ls

You should see workflow.yml listed in the output.

Don't see workflow.yml?

• Make sure you saved the file in your miniflow directory, not a parent or subfolder.
• Check that your terminal is in the miniflow folder. Run pwd to confirm.

Still stuck? Ask for help.

✔️ Awesome, I've got everything!

Great. Make sure you've saved workflow.yml.

ⓧ I'd like to double check the full code

Here is the complete workflow.yml file. Make sure yours matches exactly:

id: data_pipeline
description: A sample ETL data pipeline

tasks:
  - id: fetch_data
    type: shell
    command: "echo 'Fetching data from API...'"

  - id: validate_data
    type: python
    script: |
      import json
      data = {"users": 42, "status": "ok"}
      print(f"Validated {data['users']} records")

  - id: transform_data
    type: python
    script: |
      print("Transforming data...")
      records = [{"id": i, "value": i * 10} for i in range(5)]
      print(f"Transformed {len(records)} records")

Write the workflow parser

Your workflow file is just text on disk right now. Your engine needs a function that reads the YAML, checks it for errors, and hands back a clean Python dictionary.

• Create a new file called miniflow.py in your miniflow folder and paste the following code into it:

import yaml
import sys
import argparse
from rich.console import Console


def load_workflow(filepath):
    # Read and parse the YAML file
    with open(filepath, "r") as f:
        workflow = yaml.safe_load(f)

    # Reject empty files
    if not workflow:
        raise ValueError(f"Empty workflow file: {filepath}")

    # Verify required top-level fields exist
    required_fields = ["id", "tasks"]
    for field in required_fields:
        if field not in workflow:
            raise ValueError(f"Missing required field: '{field}' in workflow")

    # Ensure tasks is a non-empty list
    if not isinstance(workflow["tasks"], list) or len(workflow["tasks"]) == 0:
        raise ValueError("Workflow must have at least one task")

    return workflow

What does this code do?

• yaml.safe_load() parses the YAML file into a Python dictionary. The safe_load variant prevents arbitrary code execution from untrusted YAML files.
• The validation checks ensure the workflow has an id field, a tasks field, and at least one task in the list.
• If anything is wrong, the function raises a ValueError with a specific error message instead of crashing with a confusing traceback.

• Save miniflow.py.
• Test that the parser can read your workflow file by running:

python -c "from miniflow import load_workflow; wf = load_workflow('workflow.yml'); print('Loaded:', wf['id'])"

You should see Loaded: data_pipeline printed in your terminal.

Seeing a ModuleNotFoundError?

• Make sure your virtual environment is activated. You should see (venv) at the start of your terminal prompt.
• Check that miniflow.py is saved in the same directory where you're running the command.

Need more help?

The parser reads the file and checks its overall shape, but it does not yet validate individual tasks. A workflow with a missing type field or a duplicate id would slip through silently.

• In miniflow.py, find the line return workflow inside load_workflow().
• Add the following task validation code directly above return workflow:

    # Validate each task has required fields and no duplicates
    task_ids = set()
    for task in workflow["tasks"]:
        if "id" not in task:
            raise ValueError("Every task must have an 'id' field")
        if "type" not in task:
            raise ValueError(f"Task '{task['id']}' is missing a 'type' field")
        if task["type"] not in ("shell", "python"):
            raise ValueError(
                f"Task '{task['id']}' has unknown type '{task['type']}'. "
                "Supported types: shell, python"
            )
        if "command" not in task and "script" not in task:
            raise ValueError(
                f"Task '{task['id']}' must have a 'command' or 'script' field"
            )
        if task["id"] in task_ids:
            raise ValueError(f"Duplicate task id: '{task['id']}'")
        task_ids.add(task["id"])

What does this validation loop check?

The loop walks through every task in the workflow and checks five things:

• Every task has an id field so the engine can reference it.
• Every task has a type that is either shell or python.
• Every task has a command or script field so the engine knows what to execute.
• No two tasks share the same id. The task_ids set tracks every ID seen so far.

• Save miniflow.py.
• Confirm the task-level validation passes by running:

python -c "from miniflow import load_workflow; load_workflow('workflow.yml'); print('All validations passed')"

You should see All validations passed printed. If any task in your YAML had a missing field or an unsupported type, you'd see a clear error message instead.

Seeing a SyntaxError or IndentationError?

• The new code must sit inside the load_workflow() function, indented by 4 spaces. It should be directly above return workflow.
• Make sure there is no extra blank line or indentation mismatch between the new code and the existing code above it.

Can't spot the issue?

Add the CLI interface

Your parser works, but you have to import it manually in a one-line Python command every time. A proper CLI with a validate subcommand gives you a clean way to test workflows from the terminal.

• In miniflow.py, add the following main() function below the load_workflow() function. This sets up the argument parser and the validate subcommand:

def main():
    # Set up the argument parser
    parser = argparse.ArgumentParser(
        description="MiniFlow - A mini workflow orchestration engine"
    )
    subparsers = parser.add_subparsers(
        dest="command", help="Available commands"
    )

    # Register the validate subcommand
    validate_parser = subparsers.add_parser(
        "validate", help="Validate a workflow file"
    )
    validate_parser.add_argument(
        "workflow", help="Path to workflow YAML file"
    )

    args = parser.parse_args()

    if args.command is None:
        parser.print_help()
        sys.exit(1)

• Still inside main(), add the validate command handler and the entry point directly below the sys.exit(1) line:

    # Handle the validate command
    console = Console()

    if args.command == "validate":
        try:
            workflow = load_workflow(args.workflow)
            console.print(
                f"[green]Workflow '{workflow['id']}' is valid.[/green]"
            )
            console.print(f"  Tasks: {len(workflow['tasks'])}")
            for task in workflow["tasks"]:
                console.print(f"    - {task['id']} ({task['type']})")
        except (ValueError, FileNotFoundError) as e:
            console.print(f"[red]Validation error:[/red] {e}")
            sys.exit(1)


if __name__ == "__main__":
    main()

What does main() do?

• argparse handles the CLI interface. It creates a validate subcommand that expects a path to a YAML file as its argument.
• When you run python miniflow.py validate workflow.yml, it calls load_workflow() and prints each task's ID and type in green if everything passes.
• If validation fails, the error message prints in red and the script exits with code 1.
• The if __name__ == "__main__": block ensures main() only runs when you execute the file directly, not when it is imported.

• Save miniflow.py.

Before you run the validate command, what do you think the output will show for your three-task workflow?

• Test your parser and CLI by running:

python miniflow.py validate workflow.yml

You should see green text confirming the workflow is valid, along with all 3 tasks listed by ID and type:

Workflow 'data_pipeline' is valid.
  Tasks: 3
    - fetch_data (shell)
    - validate_data (python)
    - transform_data (python)

Seeing a validation error or traceback?

• If you see a FileNotFoundError, make sure workflow.yml is in the same folder as miniflow.py.
• If you see an IndentationError, check that the validate handler code is indented inside main() with 4 spaces, and that the if __name__ block is at the top level (no indentation).
• If you see no output at all, make sure you included the if __name__ == "__main__": block at the very bottom of the file.

Still stuck?

✔️ Awesome, I've got everything!

Great. Make sure you've saved miniflow.py.

ⓧ I'd like to double check the full code

Your miniflow.py has 4 sections. Scroll up and confirm each one is present in this order:

• Imports at the top: yaml, sys, argparse, and Console from Rich.
• load_workflow(): reads the YAML file, validates top-level fields, then loops through every task to check for id, type, command/script, and duplicate IDs.
• main(): sets up argparse with a validate subcommand that calls load_workflow() and prints each task.
• Entry point: if __name__ == "__main__": main() at the bottom.

Your engine can read and validate YAML workflows. Next up, you'll give it the power to actually run those tasks.

Build the Task Execution Engine

In the last step, you built a YAML parser that validates workflow definitions and checks for errors. Your engine can read a workflow file and confirm it's well-formed.

Right now, though, your orchestrator is just a YAML validator. It can tell you a workflow is valid, but it cannot actually run anything. In this step, you'll give MiniFlow the ability to execute tasks as real subprocesses and print their results.

In this step, get ready to:

• Write a run_task() function that runs shell commands and Python scripts as isolated subprocesses.
• Write an execute_workflow() function that loops through every task and prints each result to the terminal.

Write the run_task() function

Each task in your workflow is either a shell command or a Python script. Your engine needs a way to run both types in an isolated process so that a crashing script does not take down the whole orchestrator.

• In miniflow.py, find the import section at the very top of the file.
• Find this line:

import yaml
import sys
import argparse

• Add import subprocess between import yaml and import sys, so the imports now look like this:

import yaml
import subprocess
import sys
import argparse

• Now add the run_task() function by pasting the following code directly after the closing line of load_workflow() (after return workflow):

def run_task(task):
    task_type = task["type"]

    # Shell tasks run as OS commands
    if task_type == "shell":
        command = task["command"]
        result = subprocess.run(
            command, shell=True, capture_output=True, text=True
        )
        return result.returncode == 0, result.stdout, result.stderr

    # Python tasks run as scripts in a child interpreter
    elif task_type == "python":
        script = task["script"]
        result = subprocess.run(
            [sys.executable, "-c", script],
            capture_output=True,
            text=True,
        )
        return result.returncode == 0, result.stdout, result.stderr

What does this code do?

run_task() handles both task types your workflow supports. For shell tasks, it passes the command string to subprocess.run() which runs it in the operating system's shell.

For python tasks, it uses sys.executable (the path to whatever Python interpreter is running MiniFlow) and the -c flag to execute the script string in a separate child process. Both paths return a tuple of (success boolean, stdout, stderr).

Why run tasks as subprocesses?

Running each task as a subprocess gives you process isolation. If a Python script crashes or throws an unhandled exception, it only affects that child process. Your orchestrator keeps running and can report the failure cleanly.

This is the same approach production orchestrators like Kestra and Airflow use to isolate task execution from the engine itself.

Add execute_workflow() and wire the run subcommand

With run_task() ready, you need a function that loops through every task in the workflow and calls it. You also need a new run subcommand so the learner can trigger execution from the terminal.

• In miniflow.py, paste the following function directly after run_task():

def execute_workflow(workflow):
    workflow_id = workflow["id"]
    tasks = workflow["tasks"]

    # Print a header with the workflow name and task count
    print(f"Running workflow: {workflow_id}")
    print(f"Tasks: {len(tasks)}")
    print()

    # Run each task in list order and print its result
    for task in tasks:
        task_id = task["id"]
        print(f"  RUN  {task_id} ...", end="")
        success, stdout, stderr = run_task(task)
        if success:
            print(f" OK")
            if stdout.strip():
                print(f"         {stdout.strip()[:80]}")
        else:
            print(f" FAIL")
            if stderr.strip():
                print(f"         {stderr.strip()[:80]}")

    return True

What does this code do?

• The function iterates through every task in the order they appear in the YAML file.
• For each task, it prints a RUN line, calls run_task(), then prints either OK or FAIL based on the return code.
• stdout.strip()[:80] truncates each task's output to 80 characters so the terminal stays readable.

Now you need to wire this into main() so you can call it from the terminal.

• Scroll down to the main() function and find these lines:

    validate_parser = subparsers.add_parser(
        "validate", help="Validate a workflow file"
    )

• Add the following run subparser directly above those lines, so that run_parser appears before validate_parser:

    run_parser = subparsers.add_parser("run", help="Execute a workflow")
    run_parser.add_argument("workflow", help="Path to workflow YAML file")

• Now scroll to the bottom of main() and find the end of the if args.command == "validate": block. Add this elif block directly after it:

    elif args.command == "run":
        try:
            workflow = load_workflow(args.workflow)
            success = execute_workflow(workflow)
            sys.exit(0 if success else 1)
        except (ValueError, FileNotFoundError) as e:
            console.print(f"[red]Error:[/red] {e}")
            sys.exit(1)

What does this wiring do?

The run subcommand first validates the workflow with load_workflow() (catching malformed YAML before anything runs), then passes the parsed workflow to execute_workflow(). The exit code reflects whether execution succeeded, which is standard practice for CLI tools.

• Save miniflow.py.

Before you run this, what do you think will happen? Your workflow has three tasks (fetch_data, validate_data, transform_data). Will all three succeed, or will one fail?

• Run your workflow engine for the first time by running this command:

python miniflow.py run workflow.yml

You should see all three tasks execute successfully with output like this:

Running workflow: data_pipeline
Tasks: 3

  RUN  fetch_data ... OK
         Fetching data from API...
  RUN  validate_data ... OK
         Validated 42 records
  RUN  transform_data ... OK
         Transforming data...
         Transformed 5 records

Not seeing any output or getting an error?

• If you see a NameError: name 'subprocess' is not defined, check that you added import subprocess to the top of your file.
• If you see a NameError: name 'run_task' is not defined, make sure the run_task() function is defined before execute_workflow() in the file.
• If you see an indentation error, make sure the elif args.command == "run": block is at the same indentation level as the if args.command == "validate": block above it.

Still stuck? Help me debug why my miniflow.py run command is not working.

✔️ Awesome, I've got everything!

Great. Double check you've saved miniflow.py.

ⓧ I'd like to double check the full code

Your miniflow.py is now long enough that we can't show the whole file in one block. Scroll through your file and confirm these sections are present, in this order:

• Imports at the top: yaml, subprocess, sys, argparse, and Console from Rich.
• load_workflow() function (unchanged from Step 2).
• run_task() function with shell and python branches.
• execute_workflow() function that loops through tasks and prints status.
• main() with both run_parser and validate_parser subcommands, plus handlers for both.
• if __name__ == "__main__": main() at the very bottom.

Your engine can now parse and run workflows. There is a catch though. Right now, tasks execute in the order they appear in the YAML file. In the next step, you'll add dependencies between tasks and build a DAG resolver that figures out the correct execution order automatically.

Add DAG Dependency Resolution

Your engine can run all three tasks from a YAML file. It validates the workflow, executes shell commands and Python scripts, and prints the results.

There is a problem hiding in plain sight. Tasks run in whatever order you listed them in the YAML file. If a task depends on another task's output, the engine has no way to know. In this step, you will build a DAG resolver that uses topological sorting to compute the correct execution order from a dependency graph.

In this step, get ready to:

• Add dependency relationships to your workflow and observe list-order execution break.
• Build a dependency resolver using topological sort.
• Catch circular dependencies before any task runs.

Add dependencies to your workflow

Real data pipelines have tasks that depend on each other. Fetching data must finish before validation starts. Your YAML needs a way to express these relationships using a depends_on field.

• In workflow.yml, find the validate_data task.
• Add this depends_on section right after type: python and before script:

    depends_on:
      - fetch_data

• Find the transform_data task.
• Add this depends_on section right after type: python and before script:

    depends_on:
      - validate_data

Now add two new tasks to the bottom of your workflow. Notice that load_data is listed before generate_report in the file, even though load_data depends on generate_report.

• Add these two tasks at the end of the tasks list in workflow.yml, right after the transform_data task:

  - id: load_data
    type: shell
    depends_on:
      - transform_data
      - generate_report
    command: "echo 'Loading data into warehouse... Done!'"

  - id: generate_report
    type: python
    depends_on:
      - validate_data
    script: |
      print("Generating summary report...")
      print("Report: 42 users processed, 0 errors")

What does this dependency structure look like?

Your workflow now has a dependency graph. fetch_data has no dependencies. validate_data depends on fetch_data. Both transform_data and generate_report depend on validate_data.

load_data depends on both transform_data and generate_report, so it must run last. But it appears before generate_report in the file.

• Save workflow.yml.

✔️ Awesome, I've got everything!

Make sure you have saved workflow.yml with all five tasks and their depends_on fields.

ⓧ I'd like to double check the full code

Your workflow.yml is over 30 lines, so scroll through and confirm each section is present in this order:

• id: data_pipeline and description at the top.
• fetch_data (shell, no depends_on).
• validate_data (python, depends_on: fetch_data).
• transform_data (python, depends_on: validate_data).
• load_data (shell, depends_on: transform_data and generate_report).
• generate_report (python, depends_on: validate_data).

Before you run this, think about what order the five tasks will execute in. Will load_data run before or after generate_report?

• Run your workflow by running this command:

python miniflow.py run workflow.yml

load_data ran before generate_report. That is wrong. load_data depends on generate_report, so it should run after it.

The engine ran tasks in the order they appear in the YAML file. It completely ignored the depends_on fields. This is the gap you will fix next.

Build the dependency resolver

Topological sorting solves this problem. Given a graph of tasks and their dependencies, it produces an order where every task runs only after all its dependencies have finished.

What is topological sorting?

Imagine your five tasks as nodes in a graph, with arrows pointing from each task to the tasks that depend on it. Topological sorting walks this graph and produces a flat list where every task appears after its dependencies.

Python's standard library includes graphlib.TopologicalSorter for exactly this purpose. No external library needed.

• In miniflow.py, find this import line:

from rich.console import Console

• Add this new import line directly above it:

from graphlib import TopologicalSorter, CycleError

• The top of your file should now look like this:

import yaml
import subprocess
import sys
import argparse
from graphlib import TopologicalSorter, CycleError
from rich.console import Console

• Add the resolve_execution_order() function below load_workflow() and above run_task() by copying the following code:

def resolve_execution_order(tasks):
    # Build a graph mapping each task to its dependencies
    graph = {}
    for task in tasks:
        task_id = task["id"]
        dependencies = task.get("depends_on", [])
        graph[task_id] = set(dependencies)

    # Run topological sort to find a valid execution order
    try:
        sorter = TopologicalSorter(graph)
        order = list(sorter.static_order())
        return order
    except CycleError as e:
        raise ValueError(f"Circular dependency detected: {e}")

What does this code do?

The function builds a dictionary where each task ID maps to a set of its dependencies. For example, load_data maps to {"transform_data", "generate_report"}.

TopologicalSorter takes this graph and calls static_order() to produce a flat list of task IDs in a valid execution order. If it finds a cycle (a task that directly or indirectly depends on itself), it raises a CycleError.

Now you need to wire this resolver into the execution engine so it actually controls the task order.

• Find your existing execute_workflow() function in miniflow.py.
• Replace the entire function with this updated version:

def execute_workflow(workflow):
    workflow_id = workflow["id"]
    tasks = workflow["tasks"]
    # Map task IDs to task definitions for quick lookup
    task_map = {task["id"]: task for task in tasks}

    # Resolve the correct execution order from dependencies
    execution_order = resolve_execution_order(tasks)

    print(f"Running workflow: {workflow_id}")
    print(f"Tasks: {len(tasks)} | Execution order: {' -> '.join(execution_order)}")
    print()

    # Execute tasks in dependency order, not file order
    for task_id in execution_order:
        task = task_map[task_id]
        print(f"  RUN  {task_id} ...", end="")
        success, stdout, stderr = run_task(task)
        if success:
            print(f" OK")
            if stdout.strip():
                print(f"         {stdout.strip()[:80]}")
        else:
            print(f" FAIL")
            if stderr.strip():
                print(f"         {stderr.strip()[:80]}")

    return True

What changed?

task_map creates a dictionary so you can look up any task by its ID. The function calls resolve_execution_order() to get the correct order, then iterates over execution_order instead of the raw tasks list.

The print output now also shows the computed execution order, so you can see the dependency-sorted sequence at a glance.

• Save miniflow.py.
• Run your workflow again by running this command:

python miniflow.py run workflow.yml

This time, generate_report runs before load_data. The execution order line at the top confirms the dependency-sorted sequence. All five tasks show OK.

Seeing a KeyError or tasks in the wrong order?

• If you see a KeyError, check that every task ID in depends_on exactly matches an id field in your YAML. A single typo will break the lookup.
• Make sure you replaced the entire execute_workflow() function, including the for task_id in execution_order loop. If the old for task in tasks loop is still there, tasks will run in file order.

Why aren't my tasks running in the right order?

Catch circular dependencies

What happens if task A depends on task B, task B depends on task C, and task C depends on task A? No valid execution order exists. Your resolver already raises a CycleError for this case. Now you will add reference validation and update the validate command to show the computed execution order.

• In miniflow.py, find the end of the load_workflow() function. Find the return workflow line.
• Add this validation block directly above return workflow:

    # Validate that depends_on references point to existing task IDs
    for task in workflow["tasks"]:
        for dep in task.get("depends_on", []):
            if dep not in task_ids:
                raise ValueError(
                    f"Task '{task['id']}' depends on unknown task '{dep}'"
                )

What does this code do?

This loop checks every depends_on reference against the task_ids set that load_workflow() already collects. If a task says it depends on a task ID that does not exist in the workflow, the validation catches it immediately.

Next, update the validate subcommand so it also computes and displays the execution order.

• In the main() function, find the if args.command == "validate": block. Find these lines inside the try block:

            workflow = load_workflow(args.workflow)
            console.print(
                f"[green]Workflow '{workflow['id']}' is valid.[/green]"
            )
            console.print(f"  Tasks: {len(workflow['tasks'])}")
            for task in workflow["tasks"]:
                console.print(f"    - {task['id']} ({task['type']})")

• Replace those lines with this updated version:

            workflow = load_workflow(args.workflow)
            execution_order = resolve_execution_order(workflow["tasks"])
            console.print(
                f"[green]Workflow '{workflow['id']}' is valid.[/green]"
            )
            console.print(f"  Tasks: {len(workflow['tasks'])}")
            console.print(
                f"  Execution order: {' -> '.join(execution_order)}"
            )

What changed?

The validate command now calls resolve_execution_order() after loading the workflow. This means validating a file also checks for circular dependencies. The old per-task loop is replaced with a single line showing the computed execution order.

• Save miniflow.py.
• Test the updated validate command by running:

python miniflow.py validate workflow.yml

You should see a green "valid" message with the task count and the computed execution order showing the dependency-sorted sequence.

Seeing a NameError for resolve_execution_order?

• Make sure the resolve_execution_order() function is defined above main() in the file. Python must see the function definition before it is called.
• Check the from graphlib import TopologicalSorter, CycleError line is at the top of the file. If this import is missing, both TopologicalSorter and CycleError will be undefined.

Help me debug my resolve_execution_order setup.

Now create a workflow file with a circular dependency to prove the cycle detection works.

• Create a new file called workflow_cycle.yml in your miniflow project folder and paste the following code:

id: broken_pipeline
description: This workflow has a circular dependency

tasks:
  - id: task_a
    type: shell
    command: "echo 'Task A'"
    depends_on:
      - task_c

  - id: task_b
    type: shell
    command: "echo 'Task B'"
    depends_on:
      - task_a

  - id: task_c
    type: shell
    command: "echo 'Task C'"
    depends_on:
      - task_b

What is the cycle here?

task_a depends on task_c, which depends on task_b, which depends on task_a. There is no valid starting point. Every task is waiting for another task that is also waiting.

• Save workflow_cycle.yml.

✔️ Awesome, I've got everything!

Make sure you have saved workflow_cycle.yml with the three tasks forming a circular dependency.

ⓧ I'd like to double check the full code

id: broken_pipeline
description: This workflow has a circular dependency

tasks:
  - id: task_a
    type: shell
    command: "echo 'Task A'"
    depends_on:
      - task_c

  - id: task_b
    type: shell
    command: "echo 'Task B'"
    depends_on:
      - task_a

  - id: task_c
    type: shell
    command: "echo 'Task C'"
    depends_on:
      - task_b

Before you run this, what do you think will happen? Will any tasks execute, or will the engine catch the problem first?

• Validate the circular workflow by running this command:

python miniflow.py validate workflow_cycle.yml

You see a red "Circular dependency detected" error. No tasks ran. The topological sort caught the cycle during validation, before the engine tried to execute anything.

Not seeing the circular dependency error?

• Make sure the validate subcommand in main() calls resolve_execution_order(workflow["tasks"]). Without this call, the cycle check never runs during validation.
• Double check the depends_on fields in workflow_cycle.yml form a complete loop: task_a -> task_c, task_b -> task_a, task_c -> task_b.

Why isn't my circular dependency being detected?

✔️ Awesome, I've got everything!

Make sure you have saved miniflow.py with all the changes from this step.

ⓧ I'd like to double check the full code

miniflow.py is large, so scroll through and confirm each section is present in this order:

• Imports: yaml, subprocess, sys, argparse, from graphlib import TopologicalSorter, CycleError, from rich.console import Console.
• load_workflow(): reads YAML, validates fields, checks duplicates, and now validates depends_on references before return workflow.
• resolve_execution_order(): builds a dependency graph, runs TopologicalSorter, returns the sorted order.
• run_task(): unchanged from the previous step.
• execute_workflow(): uses task_map and execution_order to run tasks in dependency order.
• main(): validate subcommand now calls resolve_execution_order() and prints the execution order.

Your engine now resolves dependencies from a DAG and catches cycles before any task runs. Next up, you will replace the plain text output with a Rich-powered dashboard that tracks task state with color-coded status lines and a summary table.

Add Rich Execution Dashboard and State Tracking

Your engine resolves dependencies and executes tasks in the correct order. But every status update is plain monochrome text.

There is a bigger problem lurking underneath the surface, too. If one task fails, every downstream task still runs as if nothing happened. A real orchestrator needs to track each task's state and skip anything whose upstream dependency broke. In this step, you will add a state machine to track task lifecycle and a Rich-powered dashboard to make every state change visible at a glance.

In this step, get ready to:

• Add a TaskState enum to track task lifecycle (PENDING, RUNNING, SUCCESS, FAILED, SKIPPED).
• Replace plain print output with a Rich Console dashboard featuring a Panel header, color-coded status lines, and a summary Table.
• Add upstream failure detection that skips downstream tasks when a dependency fails.

Add the TaskState enum and new imports

Every task in the workflow will move through a series of states: it starts as PENDING, becomes RUNNING when the engine picks it up, and ends as SUCCESS or FAILED. If an upstream dependency failed, the task gets marked SKIPPED without ever running.

• In miniflow.py, replace the entire import section at the top of the file (everything above def load_workflow) with this updated version that adds the new imports and the TaskState enum:

import yaml
import subprocess
import sys
import time
import enum
import argparse
from graphlib import TopologicalSorter, CycleError
from rich.console import Console
from rich.table import Table
from rich.panel import Panel


# Define the possible states a task moves through
class TaskState(enum.Enum):
    PENDING = "PENDING"
    RUNNING = "RUNNING"
    SUCCESS = "SUCCESS"
    FAILED = "FAILED"
    SKIPPED = "SKIPPED"

What does this code do?

Three new imports join the file: time to measure how long each task takes, enum to define the state constants, and Rich's Table and Panel classes for formatted output.

The TaskState enum defines five fixed states. Using an enum instead of plain strings prevents typos and makes state comparisons explicit.

• Save miniflow.py.
• Confirm the new imports work by running this command:

python miniflow.py validate workflow.yml

You should see the same validation output as before, confirming the new imports loaded without errors.

Seeing an ImportError?

Make sure your virtual environment is still activated. You should see (venv) at the start of your terminal prompt.

Check that from rich.table import Table and from rich.panel import Panel are each on their own line with correct spelling.

Help me debug this ImportError in miniflow.py

Build the Rich display and rewrite execute_workflow

With the state enum in place, you need two things: a function that renders a summary table, and a completely rewritten execute_workflow() that tracks state, checks upstream failures, and uses Rich for color-coded output.

• In miniflow.py, add this new display_status() function directly above the existing execute_workflow() function:

def display_status(console, workflow_id, task_states, task_outputs):
    # Build a Rich table with one row per task
    table = Table(title=f"Workflow: {workflow_id}")
    table.add_column("Task", style="cyan", no_wrap=True)
    table.add_column("State", justify="center")
    table.add_column("Output", style="dim")

    state_styles = {
        TaskState.PENDING: "[dim]PENDING[/dim]",
        TaskState.RUNNING: "[yellow]RUNNING[/yellow]",
        TaskState.SUCCESS: "[green]SUCCESS[/green]",
        TaskState.FAILED: "[red]FAILED[/red]",
        TaskState.SKIPPED: "[dim]SKIPPED[/dim]",
    }

    for task_id, state in task_states.items():
        output = task_outputs.get(task_id, "")
        if output:
            output = output.strip()[:60]
        table.add_row(task_id, state_styles[state], output)

    console.print(table)

What does this function do?

display_status() creates a Rich Table with three columns: Task, State, and Output. The state_styles dictionary maps each TaskState to a color-coded Rich markup string.

Each task's output is truncated to 60 characters so the table stays readable. The engine calls this function after all tasks finish to show the final results at a glance.

• Save miniflow.py.
• Run the validate command to confirm the syntax is correct:

python miniflow.py validate workflow.yml

You should see the same valid workflow output. The new function exists but nothing calls it yet.

Now it is time to replace the old execute_workflow() function entirely. You will rebuild it in three code blocks below, each adding a section of the new function.

• Delete the entire execute_workflow() function (from def execute_workflow(workflow): down to its last line return True).
• In its place, paste this first section of the new function. It sets up state tracking and prints a startup Panel:

def execute_workflow(workflow):
    console = Console()
    workflow_id = workflow["id"]
    tasks = workflow["tasks"]
    task_map = {task["id"]: task for task in tasks}

    execution_order = resolve_execution_order(tasks)

    # Initialize every task as PENDING
    task_states = {task_id: TaskState.PENDING for task_id in execution_order}
    task_outputs = {}

    # Print a blue panel showing workflow info
    console.print(
        Panel(
            f"Starting workflow: [bold]{workflow_id}[/bold]\n"
            f"Tasks: {len(tasks)} | "
            f"Execution order: {' -> '.join(execution_order)}",
            title="MiniFlow",
            border_style="blue",
        )
    )
    console.print()

What does this section do?

task_states is a dictionary that maps every task ID to its current TaskState. Everything starts as PENDING. The task_outputs dictionary will capture stdout or stderr from each task as it runs.

The Rich Panel replaces the old plain print() header. It draws a bordered box around the workflow name and execution order, with a blue border and a "MiniFlow" title.

• Directly below console.print() (still inside execute_workflow), add the main execution loop. This checks for upstream failures before running each task:

    for task_id in execution_order:
        task = task_map[task_id]

        # Check if any upstream dependency failed
        dependencies = task.get("depends_on", [])
        upstream_failed = any(
            task_states[dep] == TaskState.FAILED for dep in dependencies
        )

        if upstream_failed:
            task_states[task_id] = TaskState.SKIPPED
            task_outputs[task_id] = "Skipped: upstream task failed"
            console.print(
                f"  [dim]SKIP[/dim]  [cyan]{task_id}[/cyan] "
                "(upstream dependency failed)"
            )
            continue

What does the upstream check do?

Before running a task, the loop checks whether any of its dependencies ended in FAILED state. If so, the task is immediately marked SKIPPED and the loop moves on with continue. This prevents wasted work and cascading errors.

• Still inside the for loop (below the continue statement), add the task execution code with color-coded output:

        task_states[task_id] = TaskState.RUNNING
        console.print(
            f"  [yellow]RUN [/yellow]  [cyan]{task_id}[/cyan] ...", end=""
        )

        # Time the task execution
        start_time = time.time()
        success, stdout, stderr = run_task(task)
        duration = time.time() - start_time

        if success:
            task_states[task_id] = TaskState.SUCCESS
            task_outputs[task_id] = stdout
            console.print(f" [green]OK[/green] ({duration:.2f}s)")
            if stdout.strip():
                console.print(f"         [dim]{stdout.strip()[:80]}[/dim]")
        else:
            task_states[task_id] = TaskState.FAILED
            task_outputs[task_id] = stderr or stdout
            console.print(f" [red]FAIL[/red] ({duration:.2f}s)")
            if stderr.strip():
                console.print(f"         [red]{stderr.strip()[:80]}[/red]")

What does this section do?

Each task transitions from RUNNING to either SUCCESS or FAILED based on the return code. Rich markup tags like [green] and [red] color the output so you can spot successes and failures instantly.

time.time() captures the duration of each task. The end="" on the RUN line keeps the cursor on the same line so the OK or FAIL result appears right next to it.

• Save miniflow.py.
• Run the workflow to see the new colored output by running this command:

python miniflow.py run workflow.yml

You should see a blue-bordered Panel at the top showing the workflow name and execution order, followed by yellow RUN lines that resolve to green OK for each task.

Seeing a SyntaxError or IndentationError?

The three code blocks you just added must all be indented inside execute_workflow(). The for loop should be at one level of indentation inside the function, and the loop body at two levels.

Make sure you deleted the entire old execute_workflow() function. Leftover lines from the old version can cause indentation conflicts.

Help me fix a SyntaxError in my rewritten execute_workflow function

Add the summary table and final counts

The engine prints colored status lines as tasks run. The last piece is a summary that tallies the results and shows a final table.

• In miniflow.py, add this final section at the bottom of execute_workflow(), after the for loop ends (at the same indentation level as the for keyword):

    console.print()
    display_status(console, workflow_id, task_states, task_outputs)

    # Count results by state
    succeeded = sum(
        1 for s in task_states.values() if s == TaskState.SUCCESS
    )
    failed = sum(
        1 for s in task_states.values() if s == TaskState.FAILED
    )
    skipped = sum(
        1 for s in task_states.values() if s == TaskState.SKIPPED
    )

    if failed == 0:
        console.print(
            f"\n[bold green]Workflow '{workflow_id}' completed successfully!"
            f"[/bold green] ({succeeded} succeeded, {skipped} skipped)"
        )
    else:
        console.print(
            f"\n[bold red]Workflow '{workflow_id}' finished with errors."
            f"[/bold red] ({succeeded} succeeded, {failed} failed, "
            f"{skipped} skipped)"
        )

    return failed == 0

What does the summary section do?

display_status() renders the Rich table you built earlier, showing every task's final state and output in a formatted grid.

The three sum() calls count how many tasks ended in each state. The final message is green if everything succeeded, or red if any task failed. return failed == 0 tells the caller (and the process exit code) whether the workflow passed.

• Save miniflow.py.

Before you run the workflow, what do you think you will see differently compared to the plain-text output from before?

• Run the full workflow by running this command:

python miniflow.py run workflow.yml

You should see a blue-bordered MiniFlow panel at the top, five color-coded task lines (yellow RUN resolving to green OK), a formatted summary table with all five tasks showing SUCCESS in green, and a bold green completion message at the bottom.

Don't see the summary table?

Make sure the summary code (starting with console.print() and display_status()) is at the same indentation level as the for loop, not inside it. If it is indented inside the loop, it will run after every single task instead of once at the end.

Check that return failed == 0 is the last line of execute_workflow(), indented one level inside the function.

Help me debug my summary table in execute_workflow

✔️ Awesome, I've got everything!

Great. Double check you have saved miniflow.py.

ⓧ I'd like to double check the full code

miniflow.py is now over 200 lines, so rather than showing it all here, scroll up and confirm each section is in place, in this order:

• The import block includes time, enum, Table, and Panel.
• The TaskState enum with five states sits below the imports, above load_workflow().
• load_workflow(), resolve_execution_order(), and run_task() are unchanged.
• display_status() sits above execute_workflow().
• execute_workflow() starts with Console + Panel, has the for loop with upstream check and color-coded output, and ends with the display_status() call, count summary, and return failed == 0.
• main() is unchanged at the bottom.

Your orchestrator now tracks every task through a full lifecycle and renders the results in a polished dashboard. Next up, you will take on a secret mission to make flaky tasks retry automatically with exponential backoff.

Secret mission

Add Retry Logic with Exponential Backoff

Real-world tasks fail. Network calls time out, services go down, and APIs return errors. Can you make MiniFlow automatically retry flaky tasks with increasing delays between attempts?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs or cloud resources to worry about. Decide whether to keep your project files, pause your environment, or delete everything entirely.

Resources you used:

• The miniflow project directory on your Desktop (contains miniflow.py, workflow.yml, workflow_cycle.yml, and your Python virtual environment).
• PyYAML and Rich libraries (installed inside your virtual environment).

Keep everything running

No action needed. Choose this if you want to keep experimenting with MiniFlow or adding new workflow files.

• Your miniflow directory stays on your Desktop with all files intact.
• Your virtual environment keeps PyYAML and Rich installed and ready to use.
• To pick up where you left off, open a terminal, navigate to the miniflow folder, and activate your virtual environment.

Pause - I'll come back to this later

Deactivate your virtual environment to free up your terminal session. Your files and installed packages stay on disk.

• Deactivate your virtual environment by running:

deactivate

• Your miniflow directory and all its files remain on your Desktop.
• To resume later, navigate to the miniflow folder and reactivate the virtual environment.

Delete - I don't want to use this again

Remove the entire project directory, including all source files, workflow definitions, and the virtual environment with its installed packages.

• Deactivate your virtual environment first by running:

deactivate

• Delete the entire miniflow project directory by running this command on macOS/Linux:

rm -rf ~/Desktop/miniflow

• On Windows, run this instead:

rmdir /s /q %USERPROFILE%\Desktop\miniflow

• This removes miniflow.py, workflow.yml, workflow_cycle.yml, and the virtual environment with PyYAML and Rich.

Nice Work!

Nice Work!

Nice work! You just built MiniFlow, a working workflow orchestration engine that reads declarative YAML pipelines and executes them with dependency resolution, state tracking, and a polished terminal UI.

You now understand the core architecture behind production tools like Kestra and Airflow.

You've learned how to:

• Build a YAML workflow parser that validates task definitions, catches duplicate IDs, and rejects unknown task types before anything runs.
• Implement DAG dependency resolution using Python's graphlib.TopologicalSorter to derive correct execution order from the dependency graph, with automatic cycle detection.
• Create a Rich-powered CLI dashboard with a state machine (PENDING, RUNNING, SUCCESS, FAILED, SKIPPED) that color-codes task execution and skips downstream tasks when an upstream dependency fails.
• Secret Mission: Added per-task retry logic with exponential backoff delay to gracefully handle flaky tasks.

Ready to quiz yourself?