Case Study: Elena's First Automated Report

The Scenario

Elena Vasquez is a data analyst at a small nonprofit called Community Bridges. Every month, the finance team sends her a CSV file of employee hours — names, departments, hours worked, and hourly rates. Her job is to produce three things:

  1. A department summary showing total hours and payroll by department
  2. A flag list of anyone who worked overtime (over 40 hours)
  3. A formatted text report that goes to the executive director's inbox

Right now, Elena does this manually. She opens the CSV in a spreadsheet, sorts it, writes formulas, copies results into an email template, and sends it off. It takes about 90 minutes every month, and she makes a mistake roughly once a quarter — usually a copy-paste error that nobody catches until payroll has already been processed.

After learning file I/O in her Python class, Elena decides to automate the whole thing.

The Data

Elena's monthly CSV looks like this (using the sample-data.csv from this chapter):

name,department,hours_worked,hourly_rate
Elena Vasquez,Programs,42,28.50
Marcus Chen,Development,38,32.00
Priya Sharma,Communications,40,26.75
David Kim,Programs,45,28.50
Sara Martinez,Development,37,32.00
James O'Brien,Finance,40,35.00
Aisha Johnson,Communications,39,26.75
Carlos Rivera,Programs,41,28.50
Lisa Nguyen,Finance,38,35.00
Tom Williams,Development,44,32.00

Elena's Solution: Step by Step

Step 1: Read and Aggregate

Elena starts by reading the CSV and computing department-level statistics:

import csv
from pathlib import Path

def read_employee_data(filepath: Path) -> list[dict]:
    """Read employee CSV into a list of dicts with computed pay."""
    employees = []
    with open(filepath, "r", newline="") as f:
        for row in csv.DictReader(f):
            row["hours_worked"] = float(row["hours_worked"])
            row["hourly_rate"] = float(row["hourly_rate"])
            row["weekly_pay"] = row["hours_worked"] * row["hourly_rate"]
            employees.append(row)
    return employees

# Test it
data_path = Path(__file__).parent / "sample-data.csv"
employees = read_employee_data(data_path)
print(f"Read {len(employees)} employee records.")

# Expected output:
# Read 10 employee records.

Step 2: Compute Department Summaries

def department_summary(employees: list[dict]) -> dict[str, dict]:
    """Aggregate stats by department."""
    depts: dict[str, dict] = {}
    for emp in employees:
        dept = emp["department"]
        if dept not in depts:
            depts[dept] = {
                "employee_count": 0,
                "total_hours": 0.0,
                "total_pay": 0.0,
            }
        depts[dept]["employee_count"] += 1
        depts[dept]["total_hours"] += emp["hours_worked"]
        depts[dept]["total_pay"] += emp["weekly_pay"]

    # Add averages
    for dept, stats in depts.items():
        stats["avg_hours"] = stats["total_hours"] / stats["employee_count"]
        stats["avg_pay"] = stats["total_pay"] / stats["employee_count"]

    return depts

summary = department_summary(employees)
for dept, stats in sorted(summary.items()):
    print(f"  {dept}: {stats['employee_count']} employees, "
          f"${stats['total_pay']:,.2f} total pay")

# Expected output:
#   Communications: 2 employees, $2,113.25 total pay
#   Development: 3 employees, $3,808.00 total pay
#   Finance: 2 employees, $2,730.00 total pay
#   Programs: 3 employees, $3,648.00 total pay

Step 3: Find Overtime Workers

def find_overtime(employees: list[dict], threshold: float = 40.0) -> list[dict]:
    """Return employees who worked more than the threshold hours."""
    return [emp for emp in employees if emp["hours_worked"] > threshold]

overtime = find_overtime(employees)
print(f"\n{len(overtime)} employees worked overtime:")
for emp in overtime:
    extra = emp["hours_worked"] - 40
    print(f"  {emp['name']}: {emp['hours_worked']} hours "
          f"({extra} hours over)")

# Expected output:
# 3 employees worked overtime:
#   Elena Vasquez: 42.0 hours (2.0 hours over)
#   David Kim: 45.0 hours (5.0 hours over)
#   Tom Williams: 44.0 hours (4.0 hours over)

Step 4: Write the Text Report

def write_report(
    employees: list[dict],
    summary: dict[str, dict],
    overtime: list[dict],
    output_path: Path,
) -> None:
    """Write a formatted text report to a file."""
    total_payroll = sum(s["total_pay"] for s in summary.values())

    with open(output_path, "w", encoding="utf-8") as f:
        f.write("=" * 50 + "\n")
        f.write("  COMMUNITY BRIDGES — MONTHLY PAYROLL REPORT\n")
        f.write("=" * 50 + "\n\n")

        # Department summary
        f.write("DEPARTMENT SUMMARY\n")
        f.write("-" * 50 + "\n")
        f.write(f"{'Department':<18} {'Staff':>5} {'Hours':>8} "
                f"{'Total Pay':>12}\n")
        f.write("-" * 50 + "\n")

        for dept in sorted(summary):
            s = summary[dept]
            f.write(f"{dept:<18} {s['employee_count']:>5} "
                    f"{s['total_hours']:>8.1f} "
                    f"${s['total_pay']:>11,.2f}\n")

        f.write("-" * 50 + "\n")
        f.write(f"{'TOTAL':<18} {len(employees):>5} "
                f"{sum(s['total_hours'] for s in summary.values()):>8.1f} "
                f"${total_payroll:>11,.2f}\n\n")

        # Overtime flag
        if overtime:
            f.write("OVERTIME ALERT\n")
            f.write("-" * 50 + "\n")
            for emp in overtime:
                extra = emp["hours_worked"] - 40
                f.write(f"  {emp['name']:<20} {emp['hours_worked']:>5.1f} hrs "
                        f"(+{extra:.1f} over)\n")
            f.write("\n")
        else:
            f.write("No overtime this period.\n\n")

        f.write("Report generated automatically by Elena's pipeline.\n")

    print(f"Report written to {output_path.name}")

report_path = Path(__file__).parent / "monthly_report.txt"
write_report(employees, summary, overtime, report_path)

Step 5: Write the Department CSV

def write_summary_csv(summary: dict[str, dict], output_path: Path) -> None:
    """Write department summary to CSV for archiving."""
    fieldnames = [
        "department", "employee_count", "total_hours",
        "total_pay", "avg_hours", "avg_pay",
    ]
    with open(output_path, "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        for dept in sorted(summary):
            row = {"department": dept}
            row.update(summary[dept])
            # Format numbers for clean CSV output
            row["total_pay"] = f"{row['total_pay']:.2f}"
            row["avg_pay"] = f"{row['avg_pay']:.2f}"
            row["total_hours"] = f"{row['total_hours']:.1f}"
            row["avg_hours"] = f"{row['avg_hours']:.1f}"
            writer.writerow(row)

    print(f"Summary CSV written to {output_path.name}")

csv_path = Path(__file__).parent / "department_summary.csv"
write_summary_csv(summary, csv_path)

The Result

Elena's automated report now runs in under a second. Here's what the output text file looks like:

==================================================
  COMMUNITY BRIDGES — MONTHLY PAYROLL REPORT
==================================================

DEPARTMENT SUMMARY
--------------------------------------------------
Department          Staff    Hours    Total Pay
--------------------------------------------------
Communications          2     79.0   $  2,113.25
Development             3    119.0   $  3,808.00
Finance                 2     78.0   $  2,730.00
Programs                3    128.0   $  3,648.00
--------------------------------------------------
TOTAL                  10    404.0   $ 12,299.25

OVERTIME ALERT
--------------------------------------------------
  Elena Vasquez           42.0 hrs (+2.0 over)
  David Kim               45.0 hrs (+5.0 over)
  Tom Williams            44.0 hrs (+4.0 over)

Report generated automatically by Elena's pipeline.

What Elena Learned

  1. Separate reading, processing, and writing. Each function does one thing. This made it easy to test each piece independently and swap out the input file.

  2. DictReader is better than reader. Accessing row["name"] is clearer than row[0], and the code doesn't break if someone adds a column to the CSV.

  3. Use pathlib for file paths. Path(__file__).parent means the script works regardless of where it's run from.

  4. Always specify encoding and newline. Elena's first version broke when a colleague saved the CSV with a different encoding. Adding encoding="utf-8" and newline="" fixed it permanently.

  5. Automation eliminates copy-paste errors. In four months of use, Elena's script has produced zero errors — compared to roughly one mistake per quarter when she did it manually.

Discussion Questions

  1. Elena's script reads the entire CSV into memory before processing it. For 10 employees, this is fine. At what point would she need to switch to line-by-line processing, and how would the code change?

  2. The executive director asks Elena to add a "month-over-month comparison" showing how each department's hours changed. What additional data would Elena need, and what file format would you recommend storing the historical data in?

  3. Elena's script doesn't handle errors — what happens if the CSV file has a missing column, an empty row, or a non-numeric value in the hours_worked column? How would you make the script more robust? (Hint: Chapter 11 covers this in detail.)

  4. A colleague asks: "Why not just use Excel?" Give two specific advantages of Elena's Python approach over doing this work in a spreadsheet.

Mini-Project

Extend Elena's pipeline to handle multiple months. Write a program that: 1. Reads all .csv files from a data/ directory (one per month) 2. Produces a separate report for each month 3. Produces a combined "quarterly summary" that shows totals across all months 4. Saves the quarterly summary as both a text report and a JSON file