Advanced Code Example — Error Handling and Debugging#
This example builds a robust data ingestion function that handles multiple exception types, logs errors to a processing report, and continues analysis even when individual records fail.
Business Scenario#
You are building a production-grade customer data processor. The input data comes from multiple sources and may contain:
- Missing required fields
- Wrong data types (text in numeric fields)
- Empty records
- Invalid value ranges
Your processor must continue through all records, collect errors, and produce a processing report at the end.
Code#
# ── Error Log ────────────────────────────────────────────────────────
processing_log = {
'processed': [],
'errors': [],
'warnings': []
}
# ── Safe Field Extractor ──────────────────────────────────────────────
def extract_numeric(value, field_name: str, customer_name: str) -> float:
"""Safely convert a value to float, raising ValueError with context."""
try:
result = float(value)
if result < 0:
raise ValueError(f"{field_name} cannot be negative: {result}")
return result
except (TypeError, ValueError) as e:
raise ValueError(f"Invalid {field_name} for {customer_name}: '{value}' — {e}")
# ── Customer Processor ────────────────────────────────────────────────
def process_customer(record: dict) -> dict | None:
"""Process a single customer record with full error handling."""
name = record.get('name', 'UNKNOWN')
try:
# Validate required fields are present
required_fields = ['name', 'total_spent', 'purchase_count', 'region']
missing = [f for f in required_fields if f not in record]
if missing:
raise KeyError(f"Missing required fields: {missing}")
# Validate and extract numeric fields
total_spent = extract_numeric(record['total_spent'], 'total_spent', name)
purchase_count = int(extract_numeric(record['purchase_count'], 'purchase_count', name))
# Business rule validation
if purchase_count == 0 and total_spent > 0:
processing_log['warnings'].append(
f"{name}: total_spent > 0 but purchase_count = 0 — possible data error"
)
# Classify tier
if total_spent >= 1000 and purchase_count >= 8:
tier = 'Platinum'
elif total_spent >= 500 or purchase_count >= 5:
tier = 'Gold'
else:
tier = 'Standard'
result = {
'name': name,
'region': record['region'],
'total_spent': total_spent,
'purchase_count': purchase_count,
'tier': tier
}
processing_log['processed'].append(name)
return result
except KeyError as e:
processing_log['errors'].append(f"{name}: Missing field — {e}")
return None
except ValueError as e:
processing_log['errors'].append(f"{name}: Data error — {e}")
return None
except Exception as e:
processing_log['errors'].append(f"{name}: Unexpected error — {type(e).__name__}: {e}")
return None
# ── Messy Input Data ──────────────────────────────────────────────────
raw_records = [
{'name': 'Alice Johnson', 'region': 'Northwest', 'total_spent': 1257.30, 'purchase_count': 12},
{'name': 'Bob Martinez', 'region': 'Southwest', 'total_spent': 'N/A', 'purchase_count': 4},
{'name': 'Carol Chen', 'region': 'Northwest', 'total_spent': 890.75, 'purchase_count': 9},
{'name': 'David Kim', 'region': 'Southeast', 'total_spent': -50.00, 'purchase_count': 2},
{'region': 'Northeast', 'total_spent': 410.50, 'purchase_count': 6}, # missing name
{'name': 'Eve Torres', 'region': 'Southwest', 'total_spent': 1450.00, 'purchase_count': 0},
{'name': 'Frank Li', 'region': 'Northeast', 'total_spent': 530.00, 'purchase_count': 7},
]
# ── Run Processing Pipeline ───────────────────────────────────────────
results = []
for record in raw_records:
outcome = process_customer(record)
if outcome:
results.append(outcome)
# ── Output Report ─────────────────────────────────────────────────────
print("=" * 58)
print(" PROCESSING RESULTS")
print("=" * 58)
for r in results:
print(f" {r['name']:<18} | {r['region']:<12} | {r['tier']}")
print("\n" + "=" * 58)
print(" PROCESSING LOG")
print("=" * 58)
print(f" Processed successfully : {len(processing_log['processed'])}")
print(f" Errors : {len(processing_log['errors'])}")
print(f" Warnings : {len(processing_log['warnings'])}")
if processing_log['errors']:
print("\n ERRORS:")
for err in processing_log['errors']:
print(f" ✗ {err}")
if processing_log['warnings']:
print("\n WARNINGS:")
for warn in processing_log['warnings']:
print(f" ⚠ {warn}")
print("=" * 58)Expected Output#
==========================================================
PROCESSING RESULTS
==========================================================
Alice Johnson | Northwest | Platinum
Carol Chen | Northwest | Gold
Eve Torres | Southwest | Standard
Frank Li | Northeast | Gold
==========================================================
PROCESSING LOG
==========================================================
Processed successfully : 4
Errors : 3
Warnings : 1
ERRORS:
✗ Bob Martinez: Data error — Invalid total_spent for Bob Martinez: 'N/A'
✗ David Kim: Data error — Invalid total_spent for David Kim: '-50.0'
✗ UNKNOWN: Missing field — Missing required fields: ['name']
WARNINGS:
⚠ Eve Torres: total_spent > 0 but purchase_count = 0 — possible data error
==========================================================Key Concepts Demonstrated#
| Concept | Where in Code |
|---|---|
try/except/except multiple handlers |
process_customer() function |
| Re-raising exceptions with context | extract_numeric() raises ValueError with message |
except Exception as e catch-all |
Last fallback in process_customer() |
| Logging to a dictionary | processing_log['errors'].append(...) |
record.get() safe access |
name = record.get('name', 'UNKNOWN') |
| List comprehension for validation | missing = [f for f in required_fields if f not in record] |
Next: Jupyter Notebook →