Python for Automation (Recap & Advanced)

import logging
from datetime import datetime
from typing import List, Dict

logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
logger = logging.getLogger(__name__)

class AutomationPipeline:
    """Robust automation pipeline with error handling."""
    
    def __init__(self, name: str):
        self.name = name
        self.steps: List[Dict] = []
        self.results: List[Dict] = []
    
    def add_step(self, name, func, **kwargs):
        self.steps.append({"name": name, "func": func, "kwargs": kwargs})
        return self
    
    def run(self):
        logger.info(f"Starting pipeline: {self.name}")
        start = datetime.now()
        
        for i, step in enumerate(self.steps, 1):
            try:
                logger.info(f"Step {i}/{len(self.steps)}: {step['name']}")
                result = step["func"](**step["kwargs"])
                self.results.append({"step": step["name"], "status": "✅", "result": result})
            except Exception as e:
                logger.error(f"Step failed: {e}")
                self.results.append({"step": step["name"], "status": "❌", "error": str(e)})
        
        elapsed = (datetime.now() - start).total_seconds()
        self.report(elapsed)
    
    def report(self, elapsed):
        print(f"\n{'='*40}")
        print(f"Pipeline: {self.name}")
        print(f"Duration: {elapsed:.2f}s")
        print(f"{'='*40}")
        for r in self.results:
            print(f"  {r['status']} {r['step']}")
        passed = sum(1 for r in self.results if r["status"] == "✅")
        print(f"\nResult: {passed}/{len(self.results)} steps passed")

# Define automation steps
def collect_data():
    return {"records": 100, "source": "database"}

def transform_data(**kwargs):
    return {"transformed": 95, "errors": 5}

def generate_report(**kwargs):
    return {"report": "report_2024.pdf"}

def send_notification(**kwargs):
    return {"sent_to": ["admin@company.com"]}

# Build and run pipeline
pipeline = AutomationPipeline("Daily Data Processing")
pipeline.add_step("Collect Data", collect_data)
pipeline.add_step("Transform Data", transform_data)
pipeline.add_step("Generate Report", generate_report)
pipeline.add_step("Send Notification", send_notification)
pipeline.run()