Cleaning CSV Data and Writing a JSON Report

Everything so far in one place: read a messy CSV, clean it, aggregate it, and write a JSON report. The script runs as-is and every step is testable.

Environment

Python 3.13, standard library only.

python -m venv .venv && source .venv/bin/activate      # Windows: .venv\Scripts\activate

The data

data/sales.csv, deliberately dirty:

date,city,amount,note
2026-10-01,Shanghai,120.50,
2026-10-01,Beijing,
2026-10-02,Shanghai,abc,amount not numeric
2026-10-02,,80.00,
2026-10-03,Beijing,200.00,

Problems to handle: empty amount, non-numeric amount, empty city, stray whitespace.

Step 1: model a row

A dataclass states the shape of a cleaned row more clearly than a dict ever will:

# clean.py
from dataclasses import dataclass, asdict
from datetime import date
from pathlib import Path

@dataclass
class Sale:
    date: date
    city: str
    amount: float

Step 2: clean one row

import csv
from collections import Counter

def parse_row(row: dict[str, str]) -> tuple[Sale | None, str | None]:
    """Clean one row. Returns (record, skip_reason); skip_reason is None on success."""
    city = (row.get("city") or "").strip()
    raw_amount = (row.get("amount") or "").strip()

    if not city:
        return None, "city is empty"
    if not raw_amount:
        return None, "amount is empty"

    try:
        amount = float(raw_amount)
    except ValueError:
        return None, f"amount is not a number: {raw_amount!r}"

    if amount < 0:
        return None, f"amount is negative: {amount}"

    return Sale(date.fromisoformat(row["date"].strip()), city, amount), None

• strip() every field and test emptiness explicitly.
• Convert with float() inside try/except ValueError; str.isdigit() cannot parse 120.50 or -5.
• Returning (result, reason) lets the caller count why rows were dropped instead of losing them silently.

Step 3: read and count

def load_sales(path: Path) -> tuple[list[Sale], Counter[str]]:
    sales: list[Sale] = []
    skipped: Counter[str] = Counter()

    with path.open(encoding="utf-8", newline="") as f:
        for row in csv.DictReader(f):
            sale, reason = parse_row(row)
            if sale is None:
                skipped[reason or "unknown"] += 1
                continue
            sales.append(sale)

    return sales, skipped

csv.DictReader gives each row as a dict keyed by the header, so column order stops mattering.

Step 4: aggregate

def summarize(sales: list[Sale]) -> dict:
    by_city: dict[str, float] = {}
    by_city_count: dict[str, int] = {}

    for sale in sales:
        by_city[sale.city] = by_city.get(sale.city, 0.0) + sale.amount
        by_city_count[sale.city] = by_city_count.get(sale.city, 0) + 1

    top_city = max(by_city, key=by_city.get) if by_city else None

    return {
        "total_records": len(sales),
        "total_amount": round(sum(s.amount for s in sales), 2),
        "average_amount": round(sum(s.amount for s in sales) / len(sales), 2) if sales else 0,
        "by_city": {city: round(amount, 2) for city, amount in sorted(by_city.items())},
        "records_by_city": dict(sorted(by_city_count.items())),
        "top_city": top_city,
    }

max(by_city, key=by_city.get) picks the city with the highest total; guard it with if by_city because an empty dict raises.

Step 5: write the report

import json

def write_report(report: dict, out_path: Path) -> None:
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(
        json.dumps(report, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

Entry point

import sys
from pathlib import Path

def main() -> int:
    base = Path(__file__).resolve().parent
    src = base / "data" / "sales.csv"
    out = base / "output" / "report.json"

    if not src.exists():
        print(f"data file not found: {src}", file=sys.stderr)
        return 1

    sales, skipped = load_sales(src)
    report = summarize(sales)
    report["skipped"] = dict(skipped)

    write_report(report, out)
    print(f"{len(sales)} records kept, {sum(skipped.values())} skipped → {out}")
    for reason, count in skipped.most_common():
        print(f"  skipped {reason}: {count}")
    return 0

if __name__ == "__main__":
    sys.exit(main())

python clean.py
# 2 records kept, 3 skipped → .../output/report.json
#   skipped amount is empty: 1
#   skipped amount is not a number: 'abc': 1
#   skipped city is empty: 1

output/report.json:

{
  "total_records": 2,
  "total_amount": 320.5,
  "average_amount": 160.25,
  "by_city": { "Beijing": 200.0, "Shanghai": 120.5 },
  "records_by_city": { "Beijing": 1, "Shanghai": 1 },
  "top_city": "Beijing",
  "skipped": { "amount is empty": 1, "amount is not a number: 'abc'": 1, "city is empty": 1 }
}

Making it testable

parse_row is a pure function, so its tests need no fixtures at all:

# tests/test_clean.py
from clean import parse_row

def test_parse_normal():
    sale, reason = parse_row({"date": "2026-10-01", "city": " Shanghai ", "amount": "120.5"})
    assert reason is None
    assert sale.city == "Shanghai"
    assert sale.amount == 120.5

def test_parse_bad_amount():
    sale, reason = parse_row({"date": "2026-10-02", "city": "Shanghai", "amount": "abc"})
    assert sale is None
    assert reason is not None

load_sales needs a real file, so build one in pytest’s tmp_path fixture (see the pytest article).

Common mistakes

1. Writing CSV without newline=""

On Windows you get a blank line between every row. Always open CSV files for writing with newline="", and also read with it so embedded newlines inside quoted fields are handled.

2. Using float for money

0.1 + 0.2 != 0.3. Use decimal.Decimal when amounts are accumulated or reconciled:

from decimal import Decimal
Decimal("120.50") + Decimal("0.25")     # Decimal('120.75')

A float is fine for display, not for accounting.

3. Dropping bad rows silently

Without a reason counter the totals never reconcile and you cannot tell why.

4. Not pinning the encoding

Write encoding="utf-8" explicitly on every open() and write_text(); the platform default is not guaranteed.

5. One function that does everything

Split parsing, aggregation and output so the parsing logic can be tested on its own.

6. Validating with isdigit()

It cannot handle decimals or a minus sign. float() plus try/except is both simpler and correct.

Summary

• Pipeline: read → parse row by row (recording skip reasons) → aggregate → write.
• Model rows with dataclass, count reasons with Counter, build paths with pathlib.
• JSON output: ensure_ascii=False plus indent=2.
• Keep parsing as pure functions and cover them with pytest.

Next: the whole Python path is complete — apply the same pattern to another language, or revisit the intermediate chapters you skipped.