🧩 Python Automation Recipes – 🏗️ Mini Automation Framework

Posted on: July 13, 2026

Description:

📌 Introduction

As automation projects grow, managing dozens of individual scripts becomes difficult. A better approach is to build a small framework where tasks are registered, configured, logged, and executed from a single entry point.

This automation recipe brings together everything you’ve learned throughout the series into a Mini Automation Framework.

It combines:

configuration
workflow execution
audit logging
retries
scheduling-ready design

This is a simplified version of the architecture used by many real-world automation platforms.

🔎 Explanation

The framework consists of four main components:

Task Registry – Stores available automation tasks.
Workflow Runner – Executes tasks in sequence.
Retry Handler – Retries failed tasks automatically.
Audit Logger – Records execution history and results.

Instead of writing one script per automation, you simply register new tasks and let the framework handle execution.

This architecture is inspired by production automation systems such as:

Apache Airflow
Prefect
Dagster
Luigi

✅ Key Takeaways

🏗️ Organize automations into a reusable framework.
🔁 Combine retries, logging, and workflow execution.
🚀 Build scalable automation systems instead of isolated scripts.

Code Snippet:

import csv
import time
from pathlib import Path
from datetime import datetime

# --- Step 1: Audit log configuration ---
AUDIT_LOG = Path("automation_audit.csv")

if not AUDIT_LOG.exists():
    with AUDIT_LOG.open("w", newline="", encoding="utf-8") as file:
        writer = csv.writer(file)
        writer.writerow(
            ["Task", "Status", "Start Time", "End Time", "Duration (s)"]
        )

# --- Step 2: Define automation tasks ---
def backup_folder():
    print("Backing up folder...")
    time.sleep(1)

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

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

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

# --- Step 4: Retry wrapper ---
def execute_task(task_name, retries=3):

    task = TASKS[task_name]

    for attempt in range(1, retries + 1):
        start = datetime.now()

        try:
            print(f"\n▶{task_name} (Attempt {attempt})")
            task()
            end = datetime.now()
            duration = (end - start).total_seconds()
            log_execution(
                task_name,
                "SUCCESS",
                start,
                end,
                duration,
            )
            print("Success")
            return

        except Exception:
            print("Failed")

            if attempt == retries:
                end = datetime.now()
                duration = (end - start).total_seconds()
                log_execution(
                    task_name,
                    "FAILED",
                    start,
                    end,
                    duration,
                )

# --- Step 5: Audit logger ---
def log_execution(task, status, start, end, duration):

    with AUDIT_LOG.open("a", newline="", encoding="utf-8") as file:

        writer = csv.writer(file)

        writer.writerow([
            task,
            status,
            start.strftime("%Y-%m-%d %H:%M:%S"),
            end.strftime("%Y-%m-%d %H:%M:%S"),
            round(duration, 2)
        ])

# --- Step 6: Workflow execution ---
workflow = [
    "backup_folder",
    "clean_logs",
    "generate_report",
]

print("Starting Automation Framework\n")

for task_name in workflow:
    execute_task(task_name)

print("\nWorkflow completed successfully.")
print(f"Audit log saved to: {AUDIT_LOG}")

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

🧩 Python Automation Recipes – 🏗️ Mini Automation Framework

Description:

📌 Introduction

🔎 Explanation

✅ Key Takeaways

Code Snippet:

Comments