From God-Function to Pipeline: Decomposing a Monolithic Data Processing Script

import csv
import os
import logging
from dataclasses import dataclass, field, asdict

import httpx

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)


# --- Data Models ---

@dataclass
class SalesRecord:
  """A single validated sales record."""
  region: str
  amount: float
  region_name: str = ""
  currency: str = "USD"
  region_total: float = 0.0


@dataclass
class PipelineStats:
  """Tracks metrics across the pipeline for observability."""
  rows_read: int = 0
  rows_dropped: int = 0
  rows_enriched: int = 0
  enrichment_errors: int = 0


# --- Stage 1: Read ---

def read_csv(file_path: str) -> list[dict]:
  """Read raw rows from a CSV file. Pure I/O — no transformation."""
  with open(file_path, "r", encoding="utf-8") as f:
      rows = list(csv.DictReader(f))
  logger.info(f"Read {len(rows)} rows from {file_path}")
  return rows


# --- Stage 2: Clean ---

def clean_record(raw: dict) -> SalesRecord | None:
  """Validate and normalize a single raw row. Returns None if invalid."""
  raw_amount = raw.get("amount", "")
  if not raw_amount:
      return None
  try:
      amount = float(raw_amount)
  except ValueError:
      return None
  if amount < 0:
      return None

  region = raw.get("region", "").strip().upper() or "UNKNOWN"
  return SalesRecord(region=region, amount=amount)


def clean_all(raw_rows: list[dict], stats: PipelineStats) -> list[SalesRecord]:
  """Clean all rows and track how many were dropped."""
  stats.rows_read = len(raw_rows)
  cleaned = []
  for row in raw_rows:
      record = clean_record(row)
      if record is not None:
          cleaned.append(record)
      else:
          stats.rows_dropped += 1
  logger.info(f"Cleaned: {len(cleaned)} valid, {stats.rows_dropped} dropped")
  return cleaned


# --- Stage 3: Aggregate ---

def compute_region_totals(records: list[SalesRecord]) -> dict[str, float]:
  """Calculate total sales per region."""
  totals: dict[str, float] = {}
  for record in records:
      totals[record.region] = totals.get(record.region, 0) + record.amount
  return totals


def attach_totals(records: list[SalesRecord], totals: dict[str, float]) -> None:
  """Mutate records in place to include their regional total."""
  for record in records:
      record.region_total = totals.get(record.region, 0.0)


# --- Stage 4: Enrich ---

def enrich_records(
  records: list[SalesRecord],
  client: httpx.Client,
  api_url: str,
  stats: PipelineStats,
) -> None:
  """Enrich each record with region metadata from an external API."""
  # Cache API responses — same region data does not change between rows
  cache: dict[str, dict] = {}
  for record in records:
      if record.region in cache:
          data = cache[record.region]
      else:
          try:
              response = client.get(f"{api_url}/regions/{record.region}")
              response.raise_for_status()
              data = response.json()
              cache[record.region] = data
              stats.rows_enriched += 1
          except (httpx.HTTPStatusError, httpx.TransportError) as e:
              logger.warning(f"Enrichment failed for region {record.region}: {e}")
              data = {}
              stats.enrichment_errors += 1

      record.region_name = data.get("name", "")
      record.currency = data.get("currency", "USD")


# --- Stage 5: Write ---

def write_csv(records: list[SalesRecord], file_path: str) -> None:
  """Write final records to a CSV file."""
  if not records:
      logger.warning("No records to write")
      return
  with open(file_path, "w", newline="", encoding="utf-8") as f:
      fieldnames = list(asdict(records[0]).keys())
      writer = csv.DictWriter(f, fieldnames=fieldnames)
      writer.writeheader()
      for record in records:
          writer.writerow(asdict(record))
  logger.info(f"Wrote {len(records)} records to {file_path}")


# --- Pipeline Orchestrator ---

def run_pipeline(input_file: str, output_file: str) -> PipelineStats:
  """Execute the full data processing pipeline in discrete stages."""
  api_url = os.environ["REGION_API_URL"]
  api_key = os.environ["REGION_API_KEY"]

  stats = PipelineStats()

  # Stage 1: Read raw data
  raw_rows = read_csv(input_file)

  # Stage 2: Clean and validate
  records = clean_all(raw_rows, stats)

  # Stage 3: Aggregate regional totals
  totals = compute_region_totals(records)
  attach_totals(records, totals)

  # Stage 4: Enrich with external data
  with httpx.Client(
      headers={"Authorization": f"Bearer {api_key}"},
      timeout=httpx.Timeout(10.0, connect=5.0),
  ) as client:
      enrich_records(records, client, api_url, stats)

  # Stage 5: Write results
  write_csv(records, output_file)

  logger.info(
      f"Pipeline complete: {stats.rows_read} read, "
      f"{stats.rows_dropped} dropped, "
      f"{stats.rows_enriched} enriched, "
      f"{stats.enrichment_errors} enrichment errors"
  )
  return stats


if __name__ == "__main__":
  run_pipeline("sales.csv", "output.csv")