Case Study: Elena's Fully Automated Report Pipeline

Contributors

Case Study: Elena's Fully Automated Report Pipeline

Elena Vasquez is a composite character based on common experiences of data analysts who learn programming. Her specific story is fictional, but the patterns she illustrates are drawn from widely reported industry experiences.

Where We Left Off

You first met Elena in Chapter 1. She was spending four hours every Friday producing the Weekly Impact Report for Harbor Community Services — downloading CSV files from three county databases, cleaning the data in Excel, writing formulas, creating charts, and emailing the results to twelve stakeholders.

In Chapter 5, she wrote her first loops to process data automatically. In Chapter 10, she learned to read and write files. In Chapter 12, she split her code into modules. In Chapter 21, she connected to a weather API for her reports. In Chapter 22, she used regex to validate data formats.

Now, in Chapter 24, Elena completes the circle. She builds a fully automated pipeline that turns her four-hour Friday ritual into a 30-second script execution — and then schedules it to run on its own.

The Final Pipeline

Elena's pipeline has five stages, each corresponding to concepts from this chapter:

Stage 1: Fetch

Elena's three county databases now publish their weekly data as downloadable CSV files at predictable URLs. Her script fetches them automatically.

import time
import requests
from pathlib import Path
from datetime import datetime

COUNTY_URLS = {
    "riverside": "https://data.riverside-county.example/weekly_services.csv",
    "lakewood": "https://data.lakewood-county.example/weekly_services.csv",
    "oakmont": "https://data.oakmont-county.example/weekly_services.csv",
}

def fetch_county_data(output_dir: Path) -> list[Path]:
    """Download CSV files from all three counties."""
    output_dir.mkdir(parents=True, exist_ok=True)
    downloaded = []

    for county, url in COUNTY_URLS.items():
        try:
            print(f"  Fetching {county}...", end=" ")
            response = requests.get(url, timeout=30)
            response.raise_for_status()

            filepath = output_dir / f"{county}_{datetime.now():%Y%m%d}.csv"
            filepath.write_text(response.text, encoding="utf-8")
            downloaded.append(filepath)
            print(f"OK ({len(response.text):,} bytes)")

            time.sleep(1)  # Polite delay between requests

        except requests.exceptions.RequestException as e:
            print(f"FAILED: {e}")

    return downloaded

What changed from her old process: Elena used to open three browser tabs, navigate to three different portals, click "Export," wait for each download, and rename the files. Now this takes three lines of execution and six seconds (including the polite delays).

Stage 2: Validate

Elena's infamous column-swap error from Chapter 1 — where she accidentally transposed two counties' data in the master spreadsheet, leading to wrong numbers at a board meeting — is now impossible. Her validation code catches structural problems before any calculations happen.

import csv

EXPECTED_COLUMNS = {"service", "clients_served", "hours_delivered", "date"}

def validate_csv(filepath: Path) -> tuple[list[dict], list[str]]:
    """Validate a county CSV file and return clean records + errors."""
    errors = []
    clean_records = []

    with open(filepath, encoding="utf-8") as f:
        reader = csv.DictReader(f)

        # Check column headers
        if not EXPECTED_COLUMNS.issubset(set(reader.fieldnames or [])):
            missing = EXPECTED_COLUMNS - set(reader.fieldnames or [])
            errors.append(f"Missing columns in {filepath.name}: {missing}")
            return [], errors

        for line_num, row in enumerate(reader, start=2):
            row_errors = []

            # Check for negative numbers
            clients = int(row.get("clients_served", 0))
            if clients < 0:
                row_errors.append(f"Negative clients_served: {clients}")

            hours = int(row.get("hours_delivered", 0))
            if hours < 0:
                row_errors.append(f"Negative hours_delivered: {hours}")

            # Check for missing service names
            if not row.get("service", "").strip():
                row_errors.append("Empty service name")

            if row_errors:
                errors.append(f"{filepath.name}, line {line_num}: "
                              + "; ".join(row_errors))
            else:
                clean_records.append({
                    "county": filepath.stem.split("_")[0],
                    "service": row["service"],
                    "clients_served": clients,
                    "hours_delivered": hours,
                    "date": row["date"],
                })

    return clean_records, errors

What changed: She used to eyeball three spreadsheets for "obvious errors." Now every row is checked against explicit rules, and problems are flagged with exact file names and line numbers.

Stage 3: Transform

The cleaned data from all three counties is merged and aggregated.

def compute_weekly_summary(records: list[dict]) -> dict:
    """Compute weekly summary statistics across all counties."""
    summary = {
        "total_clients": sum(r["clients_served"] for r in records),
        "total_hours": sum(r["hours_delivered"] for r in records),
        "by_county": {},
        "by_service": {},
    }

    for record in records:
        # Aggregate by county
        county = record["county"]
        if county not in summary["by_county"]:
            summary["by_county"][county] = {"clients": 0, "hours": 0}
        summary["by_county"][county]["clients"] += record["clients_served"]
        summary["by_county"][county]["hours"] += record["hours_delivered"]

        # Aggregate by service
        service = record["service"]
        if service not in summary["by_service"]:
            summary["by_service"][service] = {"clients": 0, "hours": 0}
        summary["by_service"][service]["clients"] += record["clients_served"]
        summary["by_service"][service]["hours"] += record["hours_delivered"]

    return summary

Stage 4: Generate Report

The summary data is formatted into a professional report — the same format the executive director expects.

def generate_report(summary: dict, errors: list[str]) -> str:
    """Generate the formatted Weekly Impact Report."""
    lines = []
    lines.append("=" * 60)
    lines.append("  HARBOR COMMUNITY SERVICES — WEEKLY IMPACT REPORT")
    lines.append(f"  Generated: {datetime.now():%Y-%m-%d %H:%M}")
    lines.append("=" * 60)
    lines.append("")
    lines.append(f"  Total Clients Served: {summary['total_clients']:>8,}")
    lines.append(f"  Total Hours Delivered: {summary['total_hours']:>8,}")
    lines.append("")

    lines.append("  BY COUNTY:")
    for county, data in sorted(summary["by_county"].items()):
        lines.append(f"    {county.title():15s}  "
                     f"Clients: {data['clients']:>6,}  "
                     f"Hours: {data['hours']:>6,}")

    lines.append("")
    lines.append("  BY SERVICE:")
    for service, data in sorted(summary["by_service"].items()):
        lines.append(f"    {service:20s}  "
                     f"Clients: {data['clients']:>6,}  "
                     f"Hours: {data['hours']:>6,}")

    if errors:
        lines.append("")
        lines.append(f"  DATA QUALITY WARNINGS ({len(errors)}):")
        for error in errors:
            lines.append(f"    ! {error}")

    lines.append("")
    lines.append("=" * 60)
    return "\n".join(lines)

Stage 5: Distribute

The report is saved to a shared drive and (conceptually) emailed to stakeholders.

import json

def save_and_distribute(report_text: str, summary: dict) -> None:
    """Save report and notify stakeholders."""
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    reports_dir = Path("weekly_reports")
    reports_dir.mkdir(exist_ok=True)

    # Save text report
    text_path = reports_dir / f"impact_report_{timestamp}.txt"
    text_path.write_text(report_text, encoding="utf-8")

    # Save JSON for programmatic access by other tools
    json_path = reports_dir / f"impact_data_{timestamp}.json"
    json_path.write_text(json.dumps(summary, indent=2), encoding="utf-8")

    print(f"  Report saved: {text_path}")
    print(f"  Data saved: {json_path}")

    # In production, Elena would also use smtplib to email the report.
    # That's beyond our scope, but the pattern is:
    #   import smtplib
    #   server = smtplib.SMTP("smtp.example.com", 587)
    #   server.send_message(msg)
    print("  (Email notification would go here)")

The Orchestrator

Elena ties it all together with a single function.

def run_weekly_report() -> None:
    """Execute the complete weekly report pipeline."""
    start = time.time()
    print("=" * 50)
    print("WEEKLY REPORT PIPELINE — Starting")
    print("=" * 50)

    # Stage 1: Fetch
    print("\n[1/5] Fetching county data...")
    data_dir = Path("data") / datetime.now().strftime("%Y%m%d")
    csv_files = fetch_county_data(data_dir)

    # Stage 2: Validate
    print("\n[2/5] Validating data...")
    all_records = []
    all_errors = []
    for csv_file in csv_files:
        records, errors = validate_csv(csv_file)
        all_records.extend(records)
        all_errors.extend(errors)
    print(f"  {len(all_records)} valid records, {len(all_errors)} errors")

    # Stage 3: Transform
    print("\n[3/5] Computing summary...")
    summary = compute_weekly_summary(all_records)

    # Stage 4: Generate
    print("\n[4/5] Generating report...")
    report = generate_report(summary, all_errors)

    # Stage 5: Distribute
    print("\n[5/5] Saving and distributing...")
    save_and_distribute(report, summary)

    elapsed = time.time() - start
    print(f"\nPipeline completed in {elapsed:.1f} seconds.")
    print("Elena's old process: ~4 hours.")
    print(f"Time saved: ~3 hours, 59 minutes, {60 - elapsed:.0f} seconds.")

Scheduling It

Elena schedules the pipeline to run every Friday at 6:00 AM, before she even gets to the office.

On her Mac at work:

crontab -e
# Add this line:
0 6 * * 5 /usr/bin/python3 /Users/elena/reports/weekly_pipeline.py >> /Users/elena/reports/pipeline.log 2>&1

On her Windows laptop at home:

schtasks /create /tn "Weekly Report" /tr "python C:\Users\Elena\reports\weekly_pipeline.py" /sc weekly /d FRI /st 06:00

The Impact

Elena reflects on her journey:

Chapter 1: She was spending 200 hours per year on a single repetitive task.
Chapter 5: She wrote her first loop and realized the computer could repeat work without getting bored.
Chapter 10: She read CSV files programmatically for the first time.
Chapter 24: The entire pipeline runs unattended. When she arrives at work on Friday morning, the report is already in her inbox — and it's never wrong.

The column-swap error that embarrassed her at the board meeting? Structurally impossible now. The validation stage catches data anomalies that she never would have spotted visually. And when a fourth county (Pinecrest) joined the program last month, she added it by changing one dictionary entry — not by redesigning her entire Excel workflow.

Discussion Questions

Robustness: What happens if one of the three county servers is down on Friday morning? How should the pipeline handle partial failures — should it generate a report with two counties' data and a warning, or refuse to run?
Monitoring: The pipeline runs at 6 AM unattended. How would Elena know if it silently fails? What monitoring would you add?
Scaling: If Harbor Community Services expands to 20 counties, what parts of Elena's pipeline would need to change? What parts would work without modification?
Validation depth: Elena's current validation checks for negative numbers and missing fields. What other validation rules would be useful for service delivery data?
The human factor: Elena's boss asks, "Can we get the report daily instead of weekly?" How much work is that change? (Hint: almost none — change the cron schedule and maybe some labels in the report.)