🧩 Python Automation Recipes – 🔗 Workflow Dependency Manager

Posted on: June 29, 2026

Description:

📌 Introduction

In real-world automation, not every task can run immediately. Some tasks depend on others to complete first.

For example:

Generate a report before emailing it.
Back up a database before cleaning old backups.
Download data before processing it.

This automation recipe shows how to execute tasks based on dependency order using Python. By defining task dependencies in a configuration, the workflow automatically determines the correct execution sequence.

🔎 Explanation

Each task defines:
- a unique name
- a list of dependencies
The script repeatedly checks which tasks are ready to execute.
A task runs only when all its dependencies have already completed.
The completed tasks are tracked until the workflow finishes.

This pattern is the foundation of:

workflow engines
CI/CD pipelines
ETL pipelines
orchestration systems

✅ Key Takeaways

🔗 Execute tasks in dependency order.
⚙️ Build smarter workflow automation.
🚀 Foundation for orchestration systems like Airflow and Prefect.

📄 Sample `workflow.json`

{
  "tasks": [
    {
      "name": "backup_database",
      "depends_on": []
    },
    {
      "name": "clean_logs",
      "depends_on": []
    },
    {
      "name": "generate_report",
      "depends_on": ["backup_database"]
    },
    {
      "name": "send_email",
      "depends_on": ["generate_report"]
    }
  ]
}

Code Snippet:

import json
import time
from pathlib import Path

# --- Step 1: Configuration ---
WORKFLOW_FILE = Path("workflow.json")

# --- Step 2: Sample task implementations ---
def backup_database():
    print("Backing up database...")
    time.sleep(1)

def clean_logs():
    print("Cleaning logs...")
    time.sleep(1)

def generate_report():
    print("Generating report...")
    time.sleep(2)

def send_email():
    print("Sending report...")
    time.sleep(1)

# --- Step 3: Task registry ---
TASKS = {
    "backup_database": backup_database,
    "clean_logs": clean_logs,
    "generate_report": generate_report,
    "send_email": send_email,
}

# --- Step 4: Load workflow configuration ---
with WORKFLOW_FILE.open("r", encoding="utf-8") as file:
    workflow = json.load(file)

pending_tasks = workflow["tasks"]
completed_tasks = set()

print("Starting workflow...\n")

# --- Step 5: Execute tasks based on dependencies ---
while pending_tasks:

    progress = False

    for task in pending_tasks[:]:

        task_name = task["name"]
        dependencies = task.get("depends_on", [])

        # Check whether all dependencies have completed
        if all(dep in completed_tasks for dep in dependencies):

            print(f"Executing: {task_name}")

            TASKS[task_name]()

            print(f"Completed: {task_name}\n")

            completed_tasks.add(task_name)
            pending_tasks.remove(task)

            progress = True

    # Detect circular or invalid dependencies
    if not progress:
        raise RuntimeError(
            "Unable to continue. Check for circular or missing dependencies."
        )

print("Workflow completed successfully.")

← →	move
↑	rotate
↓	soft drop
Space	hard drop
P	pause / resume

🧩 Python Automation Recipes – 🔗 Workflow Dependency Manager

Description:

📌 Introduction

🔎 Explanation

✅ Key Takeaways

📄 Sample `workflow.json`

Code Snippet:

Comments