Lab 04 — Functional Data Pipeline

Hands-on Lab 45 min Intermediate

Objectives

1. Write clean, documented functions with type hints
2. Use *args/**kwargs for flexible function signatures
3. Apply map(), filter(), sorted() with lambdas
4. Build a timing decorator
5. Build a factory closure to create pipeline stages

---

Step 1 — Setup

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

Create pipeline.py:

#!/usr/bin/env python3
"""
Lab 04 — Functional Data Pipeline
Process sales data using functions, lambdas, closures, and decorators.
"""

import functools
import time
from collections import defaultdict


# ── Decorators ────────────────────────────────────────────────

def timeit(func):
    """Measure and print function execution time."""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"  ⏱  {func.__name__} took {elapsed*1000:.2f}ms")
        return result
    return wrapper


def logged(func):
    """Log function calls with arguments."""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        print(f"  📝 Calling {func.__name__}")
        result = func(*args, **kwargs)
        return result
    return wrapper


# ── Data ──────────────────────────────────────────────────────

SALES_DATA = [
    {"id": 1, "product": "Laptop",  "price": 999.99, "qty": 5,  "region": "North"},
    {"id": 2, "product": "Mouse",   "price": 29.99,  "qty": 50, "region": "South"},
    {"id": 3, "product": "Monitor", "price": 349.99, "qty": 8,  "region": "North"},
    {"id": 4, "product": "Keyboard","price": 89.99,  "qty": 25, "region": "East"},
    {"id": 5, "product": "Webcam",  "price": 79.99,  "qty": 30, "region": "West"},
    {"id": 6, "product": "Headset", "price": 149.99, "qty": 15, "region": "South"},
    {"id": 7, "product": "Tablet",  "price": 599.99, "qty": 3,  "region": "East"},
    {"id": 8, "product": "Speaker", "price": 199.99, "qty": 12, "region": "North"},
]


# ── Pipeline Functions ─────────────────────────────────────────

def add_revenue(records: list[dict]) -> list[dict]:
    """Add a 'revenue' field (price × qty) to each record."""
    return [
        {**record, "revenue": round(record["price"] * record["qty"], 2)}
        for record in records
    ]


def filter_by_min_revenue(records: list[dict], minimum: float) -> list[dict]:
    """Keep only records with revenue >= minimum."""
    return list(filter(lambda r: r["revenue"] >= minimum, records))


def sort_by(records: list[dict], key: str, reverse: bool = True) -> list[dict]:
    """Sort records by a given key."""
    return sorted(records, key=lambda r: r[key], reverse=reverse)


def make_formatter(template: str):
    """Factory: returns a formatting function for a given template."""
    def format_record(record):
        return template.format(**record)
    return format_record


def group_by_region(records: list[dict]) -> dict[str, list]:
    """Group records by region using defaultdict."""
    grouped = defaultdict(list)
    for record in records:
        grouped[record["region"]].append(record)
    return dict(grouped)


def summarize(*records: dict, **options) -> dict:
    """
    Summarize a list of sales records.

    Args:
        *records: Variable number of sale dicts.
        **options: Optional filters (e.g., region='North').

    Returns:
        Summary dict with total, average, count.
    """
    region_filter = options.get("region")
    filtered = [r for r in records if not region_filter or r["region"] == region_filter]

    revenues = [r["revenue"] for r in filtered]
    if not revenues:
        return {"count": 0, "total": 0, "average": 0}

    return {
        "count":   len(revenues),
        "total":   round(sum(revenues), 2),
        "average": round(sum(revenues) / len(revenues), 2),
        "maximum": max(revenues),
        "minimum": min(revenues),
    }


# ── Main Pipeline ─────────────────────────────────────────────

@timeit
@logged
def run_pipeline(data: list[dict]) -> None:
    """Execute the full sales data pipeline."""
    print("\n" + "=" * 55)
    print(f"{'SALES DATA PIPELINE':^55}")
    print("=" * 55)

    # Stage 1: Enrich
    enriched = add_revenue(data)

    # Stage 2: Filter (revenue >= 500)
    filtered = filter_by_min_revenue(enriched, minimum=500)
    print(f"\n  Records with revenue ≥ $500: {len(filtered)}/{len(enriched)}")

    # Stage 3: Sort by revenue
    ranked = sort_by(filtered, key="revenue")

    # Stage 4: Format and display
    fmt = make_formatter("  {product:<12} ${price:>7.2f} × {qty:>3} = ${revenue:>9.2f} [{region}]")
    print(f"\n  {'Product':<12} {'Price':>8} {'Qty':>5}   {'Revenue':>10}  Region")
    print("  " + "-" * 52)
    for record in ranked:
        print(fmt(record))

    # Stage 5: Group by region
    grouped = group_by_region(enriched)
    print(f"\n  Sales by Region:")
    for region in sorted(grouped):
        region_records = grouped[region]
        total_revenue = sum(r["revenue"] for r in region_records)
        print(f"    {region:<8} — {len(region_records)} products — ${total_revenue:,.2f} total")

    # Stage 6: Summary using *args
    print(f"\n  Overall Summary:")
    summary = summarize(*enriched)
    for k, v in summary.items():
        print(f"    {k:<10}: ${v:,.2f}" if isinstance(v, float) else f"    {k:<10}: {v}")

    # Stage 7: Reduce — total revenue
    from functools import reduce
    total = reduce(lambda acc, r: acc + r["revenue"], enriched, 0)
    print(f"\n  Total revenue (reduce): ${total:,.2f}")

    print()


if __name__ == "__main__":
    run_pipeline(SALES_DATA)

---

Step 2 — Run

python3 pipeline.py

Verification ✅

  📝 Calling run_pipeline

=======================================================
                  SALES DATA PIPELINE
=======================================================

  Records with revenue ≥ $500: 5/8

  Product      Price  Qty      Revenue  Region
  ----------------------------------------------------
  Laptop     $999.99 ×   5 = $ 4999.95 [North]
  Monitor    $349.99 ×   8 = $ 2799.92 [North]
  ...

  ⏱  run_pipeline took 1.23ms

---

Bonus

Add a @retry(times=3) decorator and simulate a flaky API call with random.random() < 0.5 that raises an error 50% of the time.

---

Summary

You used:

• Type hints on all functions
• @timeit and @logged decorators with @functools.wraps
• make_formatter() closure factory returning a custom formatting function
• *records and **options in summarize()
• filter() with lambda, sorted() with lambda
• reduce() for revenue aggregation
• defaultdict(list) for grouping