Python Foundation for Data Engineers

Chapter 10 — Week 1 Checkpoint Project

Lesson 10.1 — Checkpoint: Clean, Validate, Save, Test

Here's where everything from this entire week comes together — variables, strings, collections, control flow, functions, error handling, files, and now testing — into one small, complete piece of work.

The task: you've been handed a messy customer file. Clean it, validate it, save the clean version, and prove your work with tests.

Start by creating the messy input, raw_customers.csv:

python
import csv with open("raw_customers.csv", "w", newline="") as file: writer = csv.writer(file) writer.writerow(["customer_id", "name", "email"]) writer.writerow(["101", " priya shah ", "Priya.Shah@Company.com"]) writer.writerow(["102", "raj KUMAR", "raj_kumar@@company.com"]) writer.writerow(["103", "", "amit.verma@company.com"]) writer.writerow(["104", "Neha Gupta", "neha.gupta@company.com"])

Now, in cleaning.py, build the pipeline out of pieces you already trust, because you've already tested most of them this week:

python
import csv import re def clean_name(name): return name.strip().title() def is_valid_email(email): email = email.strip().lower() has_valid_pattern = bool(re.search(r"[\w.]+@[\w.]+\.\w+", email)) has_one_at_symbol = email.count("@") == 1 return has_valid_pattern and has_one_at_symbol def clean_customer_file(input_path, output_path): with open(input_path, "r") as file: reader = csv.DictReader(file) rows = list(reader) cleaned_rows = [] skipped_count = 0 for row in rows: name = row["name"].strip() email = row["email"].strip() if name == "" or not is_valid_email(email): skipped_count += 1 continue cleaned_rows.append({ "customer_id": row["customer_id"], "name": clean_name(name), "email": email.lower(), }) with open(output_path, "w", newline="") as file: writer = csv.DictWriter(file, fieldnames=["customer_id", "name", "email"]) writer.writeheader() writer.writerows(cleaned_rows) return { "total_rows": len(rows), "cleaned_rows": len(cleaned_rows), "skipped_rows": skipped_count, }

Run it:

python
from cleaning import clean_customer_file summary = clean_customer_file("raw_customers.csv", "clean_customers.csv") print(summary)
Output / Note

{'total_rows': 4, 'cleaned_rows': 2, 'skipped_rows': 2}

Two customers cleaned successfully, two skipped — one for a missing name, one for a broken double-@ email. Open clean_customers.csv in VS Code and see the result for yourself: two properly capitalized names, two valid, lowercased emails.

Now, prove it with tests — the final step, and arguably the most important one:

python
from cleaning import clean_customer_file def test_clean_customer_file_produces_expected_counts(tmp_path): input_path = tmp_path / "raw.csv" output_path = tmp_path / "clean.csv" with open(input_path, "w", newline="") as file: writer = csv.writer(file) writer.writerow(["customer_id", "name", "email"]) writer.writerow(["1", " priya shah ", "priya.shah@company.com"]) writer.writerow(["2", "", "broken@@email.com"]) summary = clean_customer_file(str(input_path), str(output_path)) assert summary["total_rows"] == 2 assert summary["cleaned_rows"] == 1 assert summary["skipped_rows"] == 1 def test_clean_customer_file_writes_properly_formatted_output(tmp_path): input_path = tmp_path / "raw.csv" output_path = tmp_path / "clean.csv" with open(input_path, "w", newline="") as file: writer = csv.writer(file) writer.writerow(["customer_id", "name", "email"]) writer.writerow(["1", " priya shah ", "Priya.Shah@Company.com"]) clean_customer_file(str(input_path), str(output_path)) with open(output_path, "r") as file: reader = csv.DictReader(file) cleaned = list(reader) assert cleaned[0]["name"] == "Priya Shah" assert cleaned[0]["email"] == "priya.shah@company.com"
Output / Note

test_cleaning.py ........ [100%] 8 passed in 0.03s

Take a moment on what you just built. Not a toy exercise — a real, small pipeline: read messy input, clean it, validate it, write trustworthy output, and a test suite proving it actually works, on command, forever, without you ever having to eyeball the output by hand again.

That nagging feeling from the very start of this week — "did I actually do this right?" — has a real answer now. You run pytest, and it tells you, honestly, in under a second.

Try It Yourself: Add one more deliberately broken row to your test input — a customer with a name but a completely empty email — and write a test confirming clean_customer_file correctly skips it too. This closes out Week 1: you've gone from your very first print() statement to a tested, working data-cleaning pipeline, built entirely by you.