Advanced Workflow Automation

Once you're comfortable with individual scripts, the next level is chaining them into automated pipelines.

Prerequisites

• Completed at least 3 beginner tutorials
• Comfortable with Python functions, classes, and error handling
• Basic understanding of CLI tools and cron jobs

The Pipeline Pattern

A pipeline connects multiple scripts where the output of one becomes the input of the next:

Download → Validate → Transform → Store → Notify

from typing import Callable, Any


def pipeline(*steps: Callable) -> Callable:
    """Create a pipeline from a sequence of functions."""
    def run(data: Any) -> Any:
        result = data
        for step in steps:
            result = step(result)
        return result
    return run


# Usage
auto = pipeline(
    download_reports,
    validate_schema,
    transform_to_csv,
    upload_to_database,
    send_notification,
)

auto(config)

Error Recovery

Production pipelines need resilience:

import time
from functools import wraps


def retry(max_attempts: int = 3, delay: float = 1.0):
    """Retry decorator with exponential backoff."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts:
                        raise
                    wait = delay * (2 ** (attempt - 1))
                    print(f"Attempt {attempt} failed: {e}. Retrying in {wait}s...")
                    time.sleep(wait)
        return wrapper
    return decorator


@retry(max_attempts=3, delay=2.0)
def download_reports(config):
    # ... network call that might fail
    pass

Scheduling with Cron

# Run every day at 6 AM
0 6 * * * cd /path/to/project && python pipeline.py >> logs/pipeline.log 2>&1

Logging Best Practices

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.FileHandler("pipeline.log"),
        logging.StreamHandler(),
    ],
)

logger = logging.getLogger(__name__)
logger.info("Pipeline started")

What You Learned

• How to compose functions into pipelines
• Error recovery with retry and exponential backoff
• Scheduling automation with cron
• Production logging patterns