AW Dev Rethought

Truth can only be found in one place: the code - Robert C. Martin

🧩 Python Automation Recipes – ⚙️ Workflow Runner


Description:

📌 Introduction

Hardcoding every task inside a script works at first, but it becomes messy when workflows grow. A better approach is to define tasks in a config file and let Python execute them dynamically.

This automation recipe shows how to build a simple config-based workflow runner using JSON. It reads a config file, checks which tasks are enabled, and executes them in order.

This is a great foundation for:

  • automation pipelines
  • internal tools
  • batch jobs
  • mini workflow engines

🔎 Explanation

  • The workflow is defined in a separate JSON config file.
  • Each task has:
    • a task name
    • an enabled/disabled flag
  • The Python script reads the config and maps task names to functions.
  • Only enabled tasks are executed.
  • This keeps the workflow flexible without constantly changing the script.

Later, this can be extended with:

  • YAML support
  • task parameters
  • retries
  • logging
  • dependencies between tasks

✅ Key Takeaways

  • ⚙️ Run workflows from a config file.
  • 🧩 Keep automation logic clean and flexible.
  • 🚀 Foundation for building mini workflow engines.

📄 Sample workflow_config.json

{
  "workflow_name": "daily_automation",
  "tasks": [
    {
      "name": "backup_folder",
      "enabled": true
    },
    {
      "name": "clean_logs",
      "enabled": true
    },
    {
      "name": "send_report",
      "enabled": false
    }
  ]
}

Code Snippet:

import json
from pathlib import Path
from datetime import datetime

# --- Step 1: Configure config file path ---
CONFIG_FILE = Path("workflow_config.json")

# --- Step 2: Define task functions ---
def backup_folder():
    print("Running folder backup task...")
    # Add real backup logic here

def clean_logs():
    print("Running log cleanup task...")
    # Add real log cleaning logic here

def send_report():
    print("Running report email task...")
    # Add real email report logic here

# --- Step 3: Map task names from config to Python functions ---
TASK_REGISTRY = {
    "backup_folder": backup_folder,
    "clean_logs": clean_logs,
    "send_report": send_report,
}

# --- Step 4: Load workflow configuration ---
if not CONFIG_FILE.exists():
    raise FileNotFoundError(f"Config file not found: {CONFIG_FILE}")

with CONFIG_FILE.open("r", encoding="utf-8") as file:
    config = json.load(file)

workflow_name = config.get("workflow_name", "unnamed_workflow")
tasks = config.get("tasks", [])

print(f"Starting workflow: {workflow_name}")
print(f"Started at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")

# --- Step 5: Execute enabled tasks in order ---
for task in tasks:
    task_name = task.get("name")
    enabled = task.get("enabled", False)

    if not enabled:
        print(f"Skipped disabled task: {task_name}")
        continue

    task_function = TASK_REGISTRY.get(task_name)

    if task_function:
        print(f"Executing task: {task_name}")
        task_function()
        print(f"Completed task: {task_name}\n")
    else:
        print(f"Unknown task: {task_name}\n")

print("Workflow execution completed.")

Link copied!

Comments

Add Your Comment

Comment Added!