Python Tutorial / Project: Validate CSV Data and Produce Reports
A CSV reporting tool should validate its header, required fields, amounts and duplicate identifiers before aggregation. Report rejected rows and a failure exit code so that a partial report cannot be mistaken for complete data.
Define inputs, outputs and rejection policy
id identifies an expense, category names its group and amount must be a nonnegative integer written with decimal digits. The first accepted valid occurrence claims an ID; a later duplicate is rejected. This example records rejection reasons without changing the source file.
Use a fresh folder. Existing report.csv or rejected.csv causes the tool to stop. The rejected record number counts logical CSV records with the header as one; quoted multiline fields can make physical line numbers differ.
For the relevant APIs and platform requirements, see Project: Validate CSV Data and Produce Reports documentation.
Prepare the complete test input
Save expenses-input.csv with an intentional duplicate and negative amount:
id,category,amount e1,food,3500 e2,food,8000 e3,transport,2500 e2,food,8000 e4,books,-1
These defects exercise the rejection paths; they are not expected to disappear silently.
Run the complete reporting program
Save csv_report.py:
import csv
import re
from collections import defaultdict
from pathlib import Path
source = Path("expenses-input.csv")
report = Path("report.csv")
rejected = Path("rejected.csv")
if report.exists() or rejected.exists():
raise FileExistsError("Use a new output directory")
totals = defaultdict(int)
seen = set()
errors = []
accepted = 0
with source.open(encoding="utf-8-sig", newline="") as f:
reader = csv.DictReader(f)
if reader.fieldnames != ["id", "category", "amount"]:
raise ValueError("unexpected header")
for number, row in enumerate(reader, start=2):
key = (row.get("id") or "").strip()
category = (row.get("category") or "").strip()
amount_text = (row.get("amount") or "").strip()
reason = ""
if None in row or not key or not category:
reason = "invalid fields"
elif key in seen:
reason = "duplicate id"
elif not re.fullmatch(r"[0-9]+", amount_text):
reason = "invalid amount"
if reason:
errors.append({"record": number, "reason": reason})
continue
seen.add(key)
totals[category] += int(amount_text)
accepted += 1
with report.open("x", encoding="utf-8", newline="") as f:
writer = csv.writer(f)
writer.writerow(["category", "total"])
writer.writerows(sorted(totals.items()))
with rejected.open("x", encoding="utf-8", newline="") as f:
writer = csv.DictWriter(f, fieldnames=["record", "reason"])
writer.writeheader()
writer.writerows(errors)
print("accepted:", accepted, "rejected:", len(errors))
print("total:", sum(totals.values()))
raise SystemExit(1 if errors else 0)
Execute:
python csv_report.py
Expect accepted: 3 rejected: 2 and total: 14000. The exit code is 1 because rows were rejected. report.csv should contain food 11500 and transport 2500; rejected.csv should identify duplicate id and invalid amount.
Verify reports, reruns and partial-output handling
- Reconcile report totals with the valid source records.
- Correct the duplicate and negative value in a new folder, run there and verify exit code zero.
- Change the header order to inspect the strict header rule.
Creating two output files is not one atomic transaction. If the second write fails, the first file can remain. A practical pipeline may use a temporary directory, a completion marker and validation before distributing results. Do not automatically pass a partial output to downstream work.
If exporting to a spreadsheet, address formula interpretation of external strings according to the target application. The CSV format guide explains quoting and record-number limitations.
