118. 'Capstone: Csv + Re + Dataclasses'
Data cleaning pipeline using @dataclass for typed records, csv for parsing, and re for validation
118. ‘Capstone: Csv + Re + Dataclasses’
📊 A data cleaning pipeline validates and normalizes CSV student records. The email validation regex is missing anchors so partial matches pass, and the GPA calculation references the wrong field name from the CSV column headers — causing a KeyError or wrong results.
💡 Fun fact: Data cleaning is estimated to consume 60–80% of a data scientist’s time in practice. The combination of csv.DictReader, @dataclass, and re for validation is a standard “poor man’s ETL pipeline” — Extract (csv.DictReader), Transform (re validation, type conversion), Load (into typed @dataclass objects). Python’s @dataclass was inspired by the third-party attrs library and was formally added in Python 3.7 via PEP 557. Unlike namedtuples, dataclasses support mutable fields and auto-generate __repr__ and __eq__ from type-annotated class attributes.
⚠️ Watch out: re.match() without anchors only checks from the start of the string, not the end. The pattern [a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,} will match bad@@[email protected] because re.match finds @example.com as a valid end of the pattern starting somewhere in the string. You need ^ at the start and $ at the end to force the pattern to match the entire string. Alternatively, use re.fullmatch() which requires a complete match without explicit anchors.
🤔 Think about it: When the CSV column is named grade_points but the code reads row.get("gpa"), the result is None silently — not a KeyError. dict.get() returns None for missing keys. If float(None) is called, it raises TypeError; if float(row.get("gpa", 0)) is used, every GPA silently becomes 0.0. What defensive patterns would you add to catch column-name mismatches early — perhaps during the first row parsing — so the error is immediately obvious?
Learning objectives
- Use @dataclass to create typed, self-documenting record classes
- Parse CSV data with csv.DictReader and io.StringIO for in-memory strings
- Add ^ and $ anchors to regex patterns to match the full string
- Use field(default_factory=list) for mutable defaults in dataclasses
- Build a data validation pipeline that tracks errors and valid/invalid counts
Key concepts
- @dataclass — auto-generates init, repr, eq
- field(default_factory=list) — safe mutable default
- csv.DictReader(io.StringIO(s)) — parse CSV string
- ^ and $ anchors — match start and end of string
- row.get(‘column_name’) — safe dict access for CSV rows
Try it
Concept detail
Capstone: csv + re + dataclasses
@dataclass — Typed Records Without Boilerplate
from dataclasses import dataclass, field
from typing import List
@dataclass
class StudentRecord:
student_id: str
name: str
email: str
gpa: float
is_valid: bool = True # default value@dataclass auto-generates __init__, __repr__, and __eq__.
csv.DictReader — Parse CSV as Dicts
import csv, io
data = "name,score\nAlice,95\nBob,82"
reader = csv.DictReader(io.StringIO(data))
for row in reader:
print(row["name"], row["score"])
# Alice 95
# Bob 82Anchored Regex Patterns
import re
# Without anchors — matches "[email protected]" anywhere in the string:
re.match(r"[a-z]+@[a-z]+\.[a-z]+", "bad@@[email protected]") # matches!
# With anchors — must match the WHOLE string:
re.match(r"^[a-z]+@[a-z]+\.[a-z]+$", "bad@@[email protected]") # no matchfield(default_factory=list) for Mutable Defaults
@dataclass
class Report:
errors: List[str] = field(default_factory=list)
# Never use errors: List[str] = [] — shared across instances!Putting It Together
records, report = parse_students(csv_string)
valid = [r for r in records if r.is_valid]
stats = compute_stats(valid)Solution
import csv
import re
import io
from dataclasses import dataclass, field
from typing import List, Optional
@dataclass
class StudentRecord:
student_id: str
name: str
email: str
gpa: float
is_valid: bool = True
@dataclass
class CleaningReport:
total: int = 0
valid: int = 0
invalid: int = 0
errors: List[str] = field(default_factory=list)
def is_valid_email(email: str) -> bool:
"""Return True if email matches a basic email pattern."""
# FIX 1: Add ^ and $ anchors so the pattern must match the WHOLE string
pattern = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
return bool(re.match(pattern, email))
def parse_students(csv_data: str) -> tuple:
"""
Parse CSV string of student records.
CSV columns: student_id, name, email, grade_points
Returns (list of StudentRecord, CleaningReport).
"""
records = []
report = CleaningReport()
reader = csv.DictReader(io.StringIO(csv_data))
for row in reader:
report.total += 1
errors = []
email = row.get("email", "").strip()
if not is_valid_email(email):
errors.append(f"Invalid email: {email}")
# FIX 2: Read "grade_points" column (the actual CSV column name)
try:
gpa = float(row.get("grade_points", 0))
except (ValueError, TypeError):
gpa = 0.0
errors.append(f"Invalid GPA: {row.get('grade_points')}")
if gpa < 0.0 or gpa > 4.0:
errors.append(f"GPA out of range: {gpa}")
is_valid = len(errors) == 0
record = StudentRecord(
student_id=row.get("student_id", "").strip(),
name=row.get("name", "").strip(),
email=email,
gpa=gpa,
is_valid=is_valid,
)
records.append(record)
if is_valid:
report.valid += 1
else:
report.invalid += 1
report.errors.extend(errors)
return records, report
def compute_stats(records: List[StudentRecord]) -> dict:
"""Compute stats over valid records."""
valid = [r for r in records if r.is_valid]
if not valid:
return {"count": 0, "mean_gpa": 0.0, "top_student": None}
mean_gpa = sum(r.gpa for r in valid) / len(valid)
top = max(valid, key=lambda r: r.gpa)
return {"count": len(valid), "mean_gpa": round(mean_gpa, 2), "top_student": top.name}Tests
SAMPLE_CSV = """student_id,name,email,grade_points
S001,Aryan Sharma,[email protected],3.8
S002,Priya Patel,[email protected],3.5
S003,Ravi Kumar,[email protected],3.9
S004,Invalid User,notanemail,2.1
S005,Another Bad,bad@@double.com,3.0
"""
CLEAN_CSV = """student_id,name,email,grade_points
S001,Alice,[email protected],3.8
S002,Bob,[email protected],3.2
S003,Carol,[email protected],3.5
"""
def test_valid_email_accepted():
assert is_valid_email("[email protected]") is True
assert is_valid_email("[email protected]") is True
def test_invalid_email_rejected():
assert is_valid_email("notanemail") is False
assert is_valid_email("missing@") is False
def test_double_at_email_rejected():
"""bad@@double.com should fail — requires anchors to detect partial match."""
result = is_valid_email("bad@@double.com")
assert result is False, (
"bad@@double.com passed validation — add ^ and $ anchors to the regex"
)
def test_parse_students_reads_grade_points():
"""parse_students must read 'grade_points' column, not 'gpa'."""
records, report = parse_students(CLEAN_CSV)
assert len(records) == 3
# If bug is present, gpa would be 0.0 for all (row.get("gpa") returns None)
gpas = [r.gpa for r in records]
assert 3.8 in gpas, (
f"GPA 3.8 not found — are you reading the 'grade_points' column? Got: {gpas}"
)
def test_parse_students_valid_count():
records, report = parse_students(SAMPLE_CSV)
assert report.total == 5
# S001, S002, S003 are valid; S004 and S005 have bad emails
assert report.valid == 3, f"Expected 3 valid records but got {report.valid}"
assert report.invalid == 2, f"Expected 2 invalid records but got {report.invalid}"
def test_parse_returns_student_records():
records, report = parse_students(CLEAN_CSV)
assert all(isinstance(r, StudentRecord) for r in records)
assert records[0].student_id == "S001"
assert records[0].name == "Alice"
def test_compute_stats_mean_gpa():
records, _ = parse_students(CLEAN_CSV)
stats = compute_stats(records)
assert stats["count"] == 3
# mean of 3.8, 3.2, 3.5 = 3.5
assert abs(stats["mean_gpa"] - 3.5) < 0.01
def test_compute_stats_top_student():
records, _ = parse_students(CLEAN_CSV)
stats = compute_stats(records)
assert stats["top_student"] == "Alice" # highest GPA 3.8
def test_compute_stats_empty():
stats = compute_stats([])
assert stats["count"] == 0
assert stats["top_student"] is None