CSV Ingestion Validator with Custom Exceptions

What you will learn

• Design an enterprise exception hierarchy using Python class inheritance
• Isolate row-level validation errors without crashing batch pipelines
• Validate schema column completeness against required field lists
• Route malformed records into dead-letter quarantine reports

What you need first

class Base(Exception): pass
class Sub(Base): pass
print(issubclass(Sub, Base))

True

quarantine = []; quarantine.append((1, 'bad value')); print(len(quarantine))

1

val = -10.0; valid = 20.0 <= val <= 800.0; print(valid)

False

Architecture and data flow

The data

A raw CSV dataset containing patient identifiers, test dates, laboratory glucose readings, and diagnostic status flags.

Step by step

Read each step, run its code, then check your output against the one shown.

1. 01

   Define custom validation exception hierarchy

   We inspect the raw diagnostic CSV input containing a mix of valid patient records and common data entry errors. The dataset contains five rows representing blood glucose measurements across multiple hospital wards. Row two contains a negative glucose value, while row four contains an un-parsed string token 'PENDING'. In medical ingestion pipelines, parsing must anticipate malformed inputs without corrupting numerical aggregates. We inspect the raw rows to identify required schema columns and establish expected validation boundaries before implementing parsing logic and exception handling.

   PythonStep 1Copy

   class DataValidationError(Exception):
       pass
   
   class MissingColumnError(DataValidationError):
       pass
   
   class TypeConversionError(DataValidationError):
       pass
   
   class RangeError(DataValidationError):
       pass
   
   is_missing_sub = issubclass(MissingColumnError, DataValidationError)
   is_range_sub = issubclass(RangeError, DataValidationError)
   print(f"Hierarchy verified: MissingColumnError is DataValidationError: {is_missing_sub} | RangeError is DataValidationError: {is_range_sub}")

   Show outputGuess what it prints first

   Hierarchy verified: MissingColumnError is DataValidationError: True | RangeError is DataValidationError: True

   Common mistake

   Watch out: Inheriting custom exceptions from BaseException instead of Exception interferes with system signals like KeyboardInterrupt.

   How you will know

   The output confirms that derived exception classes are valid subclasses of DataValidationError.

2. 02

   Demonstrate batch termination from unhandled casting

   Wrong turn first: this step shows the mistake before the fix

   We demonstrate why standard naive parsing causes batch ETL pipelines to crash in production. When an ingestion script iterates through raw rows and attempts direct float conversion without error handling, encountering an un-parsed string like 'PENDING' raises an uncaught ValueError. In a naive pipeline, this exception terminates the Python process immediately, preventing subsequent valid patient records from being processed. This all-or-nothing failure mode causes severe operational delays and breaches healthcare delivery agreements across partner clinics. Defective rows must be quarantined instead.

   PythonStep 2Copy

   try:
       bad_val = "INVALID"
       val = float(bad_val)
   except ValueError as e:
       print("Pipeline crashed on row 2: ValueError: could not convert string to float: 'INVALID'")

   Show outputGuess what it prints first

   Pipeline crashed on row 2: ValueError: could not convert string to float: 'INVALID'

   Common mistake

   Watch out: Converting external strings to floats without try/except handling creates brittle single-point-of-failure pipelines.

   How you will know

   The output demonstrates how an unhandled ValueError halts batch execution prematurely.

3. 03

   Validate header schema completeness

   Before validating individual data values in a batch, an ingestion pipeline must verify that the file contains all mandatory header columns. In medical reporting, omitting critical fields like patient_id or glucose prevents downstream systems from linking measurements to medical histories. We define a set of required field names and compare it against the incoming CSV headers using set operations. If any required column is missing, the validator raises a MissingColumnError before attempting to process data rows, preventing confusing downstream key errors.

   PythonStep 3Copy

   required_cols = ["patient_id", "test_date", "glucose_level", "status"]
   sample_header = ["patient_id", "test_date", "glucose_level", "status"]
   
   missing = [col for col in required_cols if col not in sample_header]
   if not missing:
       print(f"Header validation passed: Required columns {required_cols} present.")

   Show outputGuess what it prints first

   Header validation passed: Required columns ['patient_id', 'test_date', 'glucose_level', 'status'] present.

   Common mistake

   Watch out: Checking column existence using string containment in raw text produces false positives if a column name is a substring of another.

   How you will know

   The check verifies that all 4 mandatory schema columns are present in the header.

4. 04

   Implement row-level validation and quarantine routing

   We construct a row validation function that applies schema and business rules sequentially. The validator first verifies that all required columns are present in the row dictionary. Next, it validates that numerical fields can be converted to float without raising ValueError. Finally, it asserts that physiological readings fall within plausible biological ranges, specifically verifying that glucose levels are strictly positive. Any violation raises the corresponding custom exception.

   PythonStep 4Copy

   import csv, io
   
   class DataValidationError(Exception): pass
   class TypeConversionError(DataValidationError): pass
   class RangeError(DataValidationError): pass
   
   raw_csv = """patient_id,test_date,glucose_level,status
   P101,2024-03-01,110.5,NORMAL
   P102,2024-03-01,-15.0,ERROR
   P103,2024-03-02,142.0,ELEVATED
   P104,2024-03-02,PENDING,UNKNOWN
   P105,2024-03-03,98.2,NORMAL
   """
   
   reader = list(csv.DictReader(io.StringIO(raw_csv.strip())))
   valid = []
   quarantine = []
   
   for idx, r in enumerate(reader, start=1):
       raw_glucose = r["glucose_level"]
       try:
           try:
               val = float(raw_glucose)
           except ValueError:
               raise TypeConversionError(f"Cannot cast '{raw_glucose}' to numeric glucose value")
           if not (20.0 <= val <= 800.0):
               raise RangeError(f"Glucose value {val} below minimum safe limit 20.0")
           valid.append(r)
       except DataValidationError as err:
           quarantine.append((idx, type(err).__name__, str(err)))
   
   print(f"Validated {len(reader)} rows: {len(valid)} valid records, {len(quarantine)} quarantined errors.")

   Show outputGuess what it prints first

   Validated 5 rows: 3 valid records, 2 quarantined errors.

   Common mistake

   Watch out: Discarding error diagnostic strings during quarantine prevents upstream clinic technicians from knowing what caused the rejection.

   How you will know

   The output confirms 3 valid records retained and 2 malformed records routed to quarantine.

5. 05

   Format ingestion summary and quarantine audit log

   Finally, we implement a resilient batch ingestion loop that processes patient records with row-level error isolation. For each row in the input dataset, we execute the validator inside a try-except block catching DataValidationError. Valid records are committed to a clean records list, while defective rows are diverted to a quarantine log capturing the patient ID, row index, and specific failure reason. We print an executive summary showing throughput counts and quarantine details to provide operators with immediate visibility into data quality.

   PythonStep 5Copy

   import csv, io
   
   class DataValidationError(Exception): pass
   class TypeConversionError(DataValidationError): pass
   class RangeError(DataValidationError): pass
   
   raw_csv = """patient_id,test_date,glucose_level,status
   P101,2024-03-01,110.5,NORMAL
   P102,2024-03-01,-15.0,ERROR
   P103,2024-03-02,142.0,ELEVATED
   P104,2024-03-02,PENDING,UNKNOWN
   P105,2024-03-03,98.2,NORMAL
   """
   
   reader = list(csv.DictReader(io.StringIO(raw_csv.strip())))
   valid = []
   quarantine = []
   
   for idx, r in enumerate(reader, start=1):
       raw_glucose = r["glucose_level"]
       try:
           try:
               val = float(raw_glucose)
           except ValueError:
               raise TypeConversionError(f"Cannot cast '{raw_glucose}' to numeric glucose value")
           if not (20.0 <= val <= 800.0):
               raise RangeError(f"Glucose value {val} below minimum safe limit 20.0")
           valid.append(r)
       except DataValidationError as err:
           quarantine.append((idx, type(err).__name__, str(err)))
   
   print("=== Ingestion Summary ===")
   print(f"Total Processed: {len(reader)}")
   print(f"Valid Records: {len(valid)} ({round(len(valid) * 100 / len(reader), 1)}%)")
   print(f"Quarantined: {len(quarantine)} ({round(len(quarantine) * 100 / len(reader), 1)}%)")
   print("Quarantine Log:")
   for row_idx, err_name, msg in quarantine:
       print(f"  Row {row_idx}: {err_name}: {msg}")

   Show outputGuess what it prints first

   === Ingestion Summary ===
   Total Processed: 5
   Valid Records: 3 (60.0%)
   Quarantined: 2 (40.0%)
   Quarantine Log:
     Row 2: RangeError: Glucose value -15.0 below minimum safe limit 20.0
     Row 4: TypeConversionError: Cannot cast 'PENDING' to numeric glucose value

   Common mistake

   Watch out: Omitting total processed counts obscures whether a 100% valid rate represents 1 row or 10,000 rows.

   How you will know

   The formatted summary prints 3 valid records (60.0%), 2 quarantined (40.0%), and lists both error lines.

Build it yourself

Write your solution below, run it in your browser, and pass every check.

What the result means

Interpretation

Out of 5 patient records processed, 3 records (60.0%) satisfied all physiological and schema invariants, while 2 records were successfully quarantined due to a negative reading (-15.0) and an un-parsed string ('PENDING'). Rather than aborting the entire batch and withholding reports for the 3 healthy patients, the system ingested valid records immediately and generated an itemized error log for laboratory remediation.

The message you would send

Hi Clinical Operations, the overnight diagnostic ingestion job processed 5 patient lab records with a 60% clean throughput. 3 valid patient reports have been committed to the warehouse, while 2 malformed records (Row 2 negative glucose, Row 4 non-numeric text) were diverted to quarantine. The remediation log has been dispatched to partner labs.

Put it on your CV

Resume or LinkedIn bullet

Engineered a resilient batch ingestion validator in Python with a custom domain exception hierarchy, routing malformed laboratory records to dead-letter quarantine without halting pipeline throughput.

README for GitHub

# CSV Ingestion Validator with Custom Exceptions

## Business Context
Nightly patient diagnostic ingestion crashed abruptly upon encountering non-numeric test strings, delaying critical lab results and breaching healthcare SLAs.

## Solution Architecture
- Structured a custom exception tree rooted in `DataValidationError` with specific derived exception classes.
- Implemented row-level exception trapping to route defects into an append-only quarantine audit log without crashing the batch.
- Produced executive reconciliation summaries detailing valid throughput and itemized defect causes.

## Verification
Passes all automated verification checks on Python 3 standard library.

Keep going

• LessonData Validation and Schema Enforcement
• Case studyRoute malformed bronze lakehouse records to a quarantine dead-letter table
• Case studyClean raw user sign-ups and validate contact records in Python