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#

ConceptWhere in Code
try/except/except multiple handlersprocess_customer() function
Re-raising exceptions with contextextract_numeric() raises ValueError with message
except Exception as e catch-allLast fallback in process_customer()
Logging to a dictionaryprocessing_log['errors'].append(...)
record.get() safe accessname = record.get('name', 'UNKNOWN')
List comprehension for validationmissing = [f for f in required_fields if f not in record]

Next: Jupyter Notebook →