Lab 07 — CSV/JSON Data Processing Pipeline

Hands-on Lab 45 min Intermediate

Objectives

1. Read a CSV dataset with csv.DictReader
2. Process and transform data in memory
3. Write filtered results to a new CSV
4. Export a summary report as JSON
5. Use pathlib for all file paths

---

Step 1 — Setup

mkdir ~/python-course/lab-07 && cd ~/python-course/lab-07

Create employees.csv:

id,name,department,salary,years,city
1,Alice Martin,Engineering,95000,5,Montreal
2,Bob Nguyen,Marketing,72000,3,Toronto
3,Charlie Smith,Engineering,88000,7,Vancouver
4,Diana Tremblay,HR,65000,2,Montreal
5,Eve Wilson,Engineering,102000,10,Toronto
6,Frank Dubois,Marketing,78000,4,Montreal
7,Grace Kim,Engineering,91000,6,Vancouver
8,Henry Lavoie,HR,58000,1,Toronto
9,Isabelle Roy,Engineering,97000,8,Montreal
10,Jack Thompson,Marketing,85000,5,Vancouver

---

Step 2 — Build the Pipeline

Create pipeline.py:

#!/usr/bin/env python3
"""Lab 07 — CSV/JSON Data Pipeline"""

import csv
import json
from pathlib import Path
from collections import defaultdict
from contextlib import contextmanager


DATA_DIR = Path(__file__).parent
INPUT_CSV = DATA_DIR / "employees.csv"
OUTPUT_CSV = DATA_DIR / "senior_engineers.csv"
REPORT_JSON = DATA_DIR / "salary_report.json"


@contextmanager
def safe_open(path, mode="r", **kwargs):
    """Context manager with error handling."""
    try:
        with open(path, mode, **kwargs) as f:
            yield f
    except FileNotFoundError:
        print(f"❌ File not found: {path}")
        yield None


def load_employees(path: Path) -> list[dict]:
    """Load employees from CSV, converting types."""
    employees = []
    with safe_open(path, newline="", encoding="utf-8") as f:
        if f is None:
            return []
        for row in csv.DictReader(f):
            employees.append({
                "id":         int(row["id"]),
                "name":       row["name"],
                "department": row["department"],
                "salary":     float(row["salary"]),
                "years":      int(row["years"]),
                "city":       row["city"],
            })
    return employees


def filter_senior_engineers(employees, min_years=5, min_salary=90000):
    """Return engineers with 5+ years and salary >= 90000."""
    return [
        e for e in employees
        if e["department"] == "Engineering"
        and e["years"] >= min_years
        and e["salary"] >= min_salary
    ]


def save_csv(records: list[dict], path: Path):
    """Save records to CSV."""
    if not records:
        return
    with open(path, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=records[0].keys())
        writer.writeheader()
        writer.writerows(records)
    print(f"  ✅ Saved {len(records)} records to {path.name}")


def build_salary_report(employees: list[dict]) -> dict:
    """Build a salary report grouped by department."""
    by_dept = defaultdict(list)
    for emp in employees:
        by_dept[emp["department"]].append(emp["salary"])

    report = {
        "total_employees": len(employees),
        "departments": {}
    }
    for dept, salaries in sorted(by_dept.items()):
        report["departments"][dept] = {
            "count":   len(salaries),
            "average": round(sum(salaries) / len(salaries), 2),
            "min":     min(salaries),
            "max":     max(salaries),
            "total":   sum(salaries),
        }
    return report


def save_json(data: dict, path: Path):
    """Save data as formatted JSON."""
    with open(path, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2, ensure_ascii=False)
    print(f"  ✅ Saved report to {path.name}")


def main():
    print("=" * 50)
    print("  CSV/JSON Data Pipeline — Lab 07")
    print("=" * 50)

    # Load
    employees = load_employees(INPUT_CSV)
    print(f"\n  Loaded {len(employees)} employees")

    # Filter
    senior_eng = filter_senior_engineers(employees)
    print(f"  Senior engineers: {len(senior_eng)}")
    for e in sorted(senior_eng, key=lambda x: x["salary"], reverse=True):
        print(f"    {e['name']:<20} ${e['salary']:>10,.0f}  ({e['years']} yrs)")

    # Save filtered CSV
    save_csv(senior_eng, OUTPUT_CSV)

    # Build and save JSON report
    report = build_salary_report(employees)
    save_json(report, REPORT_JSON)

    # Preview JSON
    print(f"\n  📊 Salary Report Preview:")
    print(json.dumps(report["departments"], indent=4))


if __name__ == "__main__":
    main()

---

Step 3 — Run

python3 pipeline.py

Verification ✅

==================================================
  CSV/JSON Data Pipeline — Lab 07
==================================================

  Loaded 10 employees
  Senior engineers: 3
    Eve Wilson           $102,000  (10 yrs)
    Isabelle Roy          $97,000  (8 yrs)
    Alice Martin          $95,000  (5 yrs)
  ✅ Saved 3 records to senior_engineers.csv
  ✅ Saved report to salary_report.json

---

Summary

• csv.DictReader + type conversion for clean data loading
• Custom @contextmanager (safe_open) for error-handled file operations
• pathlib.Path for all file paths — clean and cross-platform
• defaultdict(list) for grouping by department
• json.dump(..., indent=2) for readable JSON output