Python Error Handling: try/except/finally, Exception Hierarchy, Custom Exceptions, Raising Errors, and Every Pattern Data Engineers Need

Every data pipeline will fail. A file will be missing. A database connection will time out. An API will return garbage instead of JSON. A CSV will have a row with five columns instead of six. The question is not if your code will encounter errors — it is how your code will respond when it does.

Error handling is what separates a script that crashes at 3 AM and wakes up the on-call engineer from a pipeline that logs the error, retries the operation, skips the bad record, sends a Slack alert, and keeps running.

Real-life analogy: Think of error handling like a hospital emergency room. The ER does not shut down when a patient arrives with a broken arm. It triages the situation (identify the error type), treats the specific injury (handle the specific exception), documents everything (log the error), and keeps operating for the next patient. A program without error handling is like a hospital that locks its doors after the first emergency. A program with good error handling triages every situation, treats what it can, and escalates what it cannot.

Table of Contents

• What Are Exceptions?
• Syntax Errors vs Exceptions
• Common Built-in Exceptions
• The Exception Hierarchy
• Basic try/except
• Your First try/except
• Catching Specific Exceptions
• Catching Multiple Exceptions
• Catching the Exception Message
• The else Clause
• The finally Clause
• Complete try/except/else/finally
• Why finally Matters for Data Engineering
• Raising Exceptions
• raise — Throw Your Own Errors
• Re-raising Exceptions
• Raise with Custom Messages
• Custom Exceptions
• Why Custom Exceptions
• Creating Custom Exceptions
• Custom Exceptions with Extra Attributes
• Exception Groups (Data Pipeline Pattern)
• Nested try/except
• Exception Chaining (from keyword)
• Common Patterns for Data Engineers
• Retry with Exponential Backoff
• Skip Bad Records and Log
• Validate Before Processing
• Database Connection Safety
• API Error Handling
• File Processing with Error Recovery
• Anti-Patterns — What NOT to Do
• Best Practices Summary
• Common Mistakes
• Interview Questions
• Wrapping Up

What Are Exceptions?

Syntax Errors vs Exceptions

# SYNTAX ERROR — Python cannot even parse the code
# These happen BEFORE the code runs — you cannot catch them with try/except
if True
    print("hello")   # SyntaxError: expected ':'

# EXCEPTION — Code is valid Python, but something goes wrong at runtime
# These happen DURING execution — you CAN catch them with try/except
x = 10 / 0           # ZeroDivisionError: division by zero
int("hello")          # ValueError: invalid literal for int()
my_list = [1, 2, 3]
my_list[10]           # IndexError: list index out of range

# Key difference:
# Syntax errors = your code is written wrong (fix the code)
# Exceptions = your code is written correctly but hit a runtime problem (handle it)

Common Built-in Exceptions

Exception	When It Happens	Example
FileNotFoundError	File does not exist	open("missing.csv")
KeyError	Dict key does not exist	config["missing_key"]
IndexError	List index out of range	my_list[99]
TypeError	Wrong type for operation	"age: " + 25
ValueError	Right type, wrong value	int("abc")
ZeroDivisionError	Dividing by zero	100 / 0
AttributeError	Object has no such attribute	None.upper()
ImportError	Module not found	import nonexistent
ConnectionError	Network connection failed	Database timeout
TimeoutError	Operation timed out	API call too slow
PermissionError	No access to resource	Reading a protected file
UnicodeDecodeError	Wrong file encoding	UTF-8 file read as ASCII
StopIteration	Iterator exhausted	next() on empty iterator

The Exception Hierarchy

# All exceptions inherit from BaseException
# Most exceptions you handle inherit from Exception
#
# BaseException
#  ├── SystemExit              ← sys.exit() — do NOT catch this
#  ├── KeyboardInterrupt       ← Ctrl+C — do NOT catch this
#  ├── GeneratorExit           ← generator cleanup — do NOT catch this
#  └── Exception               ← CATCH FROM HERE
#       ├── ValueError
#       ├── TypeError
#       ├── KeyError
#       ├── IndexError
#       ├── FileNotFoundError  (subclass of OSError)
#       ├── ConnectionError    (subclass of OSError)
#       ├── TimeoutError       (subclass of OSError)
#       ├── PermissionError    (subclass of OSError)
#       ├── UnicodeDecodeError (subclass of ValueError)
#       └── ... hundreds more

# WHY THIS MATTERS:
# except Exception catches everything you should handle
# except BaseException catches SystemExit and KeyboardInterrupt too — DANGEROUS
# Always catch Exception, never BaseException

# Check if one exception is a subclass of another:
print(issubclass(FileNotFoundError, OSError))     # True
print(issubclass(FileNotFoundError, Exception))   # True
print(issubclass(KeyboardInterrupt, Exception))   # False — it is under BaseException

Basic try/except

Your First try/except

# Without error handling — program crashes
result = 10 / 0   # ZeroDivisionError — program stops here, everything after is lost

# With error handling — program continues
try:
    result = 10 / 0
except ZeroDivisionError:
    print("Cannot divide by zero!")
    result = 0

print(f"Result: {result}")  # Result: 0 — program continues running

# The structure:
# try:        ← "Try to run this code"
#     ...     ← The code that MIGHT fail
# except:     ← "If it fails, run this instead"
#     ...     ← The recovery code

Catching Specific Exceptions

# ALWAYS catch specific exceptions — never bare except
# Each exception type gets its own recovery logic

# File reading
try:
    with open("data.csv", "r") as f:
        content = f.read()
except FileNotFoundError:
    print("Data file not found — using empty dataset")
    content = ""

# Dictionary access
config = {"host": "localhost"}
try:
    port = config["port"]
except KeyError:
    print("Port not in config — using default 5432")
    port = 5432

# Type conversion
raw_value = "not_a_number"
try:
    value = int(raw_value)
except ValueError:
    print(f"Cannot convert '{raw_value}' to int — using 0")
    value = 0

Catching Multiple Exceptions

# Multiple except blocks — different handling for each type
try:
    with open("config.json", "r") as f:
        config = json.load(f)
    db_host = config["db_host"]
    port = int(config["port"])
except FileNotFoundError:
    print("Config file missing — using defaults")
    config = {"db_host": "localhost", "port": 5432}
except json.JSONDecodeError:
    print("Config file is not valid JSON — using defaults")
    config = {"db_host": "localhost", "port": 5432}
except KeyError as e:
    print(f"Missing config key: {e} — using defaults")
    config = {"db_host": "localhost", "port": 5432}

# Catch multiple exceptions in ONE block (same handling)
try:
    value = int(user_input)
except (ValueError, TypeError):
    print("Invalid input — must be a number")
    value = 0

# Order matters — catch SPECIFIC exceptions first, GENERAL last
try:
    data = process(file)
except FileNotFoundError:
    print("File not found")        # Specific — caught first
except OSError:
    print("OS error occurred")     # General — catches remaining OS errors
except Exception:
    print("Something else failed") # Most general — catches everything else

Catching the Exception Message

# Use "as e" to capture the exception object
try:
    with open("missing.csv", "r") as f:
        data = f.read()
except FileNotFoundError as e:
    print(f"Error: {e}")
    # Error: [Errno 2] No such file or directory: 'missing.csv'

# The exception object contains:
# str(e)       — human-readable message
# type(e)      — the exception class
# e.args       — tuple of arguments passed to the exception
# e.__traceback__ — the traceback object (for logging)

try:
    result = int("hello")
except ValueError as e:
    print(f"Type: {type(e).__name__}")   # Type: ValueError
    print(f"Message: {e}")               # Message: invalid literal for int() with base 10: 'hello'
    print(f"Args: {e.args}")             # Args: ("invalid literal for int() with base 10: 'hello'",)

The else Clause

# else runs ONLY if try succeeds (no exception occurred)
# Use it to separate "risky code" from "success code"

try:
    with open("data.csv", "r") as f:
        content = f.read()
except FileNotFoundError:
    print("File not found — skipping")
else:
    # This runs ONLY if the file was read successfully
    row_count = content.count("
")
    print(f"File loaded: {row_count} rows")

# WHY use else instead of putting code in try?
# Because code in else is NOT protected by except.
# If row_count calculation fails, you WANT it to crash loudly
# (it is a bug in YOUR code, not an expected file error).

# Without else — mask bugs:
try:
    content = open("data.csv").read()
    row_count = content.count("
")   # Bug here? Silently caught below!
except FileNotFoundError:
    print("File not found")           # This catches file errors AND bugs above

# With else — bugs in success code crash loudly (correct behavior):
try:
    content = open("data.csv").read()
except FileNotFoundError:
    print("File not found")
else:
    row_count = content.count("
")   # Bug here? Crashes normally — as it should

The finally Clause

# finally ALWAYS runs — whether try succeeds, except catches, or error propagates
# Use it for cleanup: closing connections, releasing locks, writing logs

try:
    connection = database.connect()
    result = connection.execute("SELECT * FROM customers")
except ConnectionError:
    print("Database connection failed")
except TimeoutError:
    print("Query timed out")
finally:
    # This runs NO MATTER WHAT — even if an unhandled exception occurs
    connection.close()
    print("Connection closed")

# finally even runs if you RETURN from inside try:
def get_data():
    try:
        return load_file("data.csv")
    except FileNotFoundError:
        return []
    finally:
        print("Cleanup done")  # STILL RUNS even after return

get_data()  # "Cleanup done" is printed after the return

Complete try/except/else/finally

# The complete structure — all four clauses together
import json

def load_pipeline_config(filepath):
    """Load and validate a pipeline config file."""
    try:
        # RISKY CODE — might fail
        with open(filepath, "r") as f:
            config = json.load(f)
    except FileNotFoundError:
        # HANDLE file not found
        print(f"Config file not found: {filepath}")
        return None
    except json.JSONDecodeError as e:
        # HANDLE invalid JSON
        print(f"Invalid JSON in {filepath}: {e}")
        return None
    else:
        # SUCCESS — file loaded and parsed
        print(f"Config loaded: {len(config)} keys")
        return config
    finally:
        # ALWAYS — cleanup or logging
        print(f"Config load attempt complete for: {filepath}")

# Execution flow:
# 1. try block runs
# 2. If exception → matching except block runs → finally runs
# 3. If no exception → else block runs → finally runs
# 4. finally ALWAYS runs last

Why finally Matters for Data Engineering

Real-life analogy: finally is like the “close-up checklist” in a commercial kitchen. Whether the dinner service was perfect or the kitchen caught fire, you still turn off the gas, lock the doors, and set the alarm. The kitchen cannot be left in an unsafe state regardless of what happened during service. In data engineering, finally ensures database connections are closed, temp files are deleted, and audit logs are written — no matter what went wrong.

# Without finally — connection leak
def query_database(sql):
    conn = db.connect()
    try:
        result = conn.execute(sql)
        return result
    except Exception:
        print("Query failed")
        # conn.close() is NEVER called if we return early or crash!

# With finally — guaranteed cleanup
def query_database(sql):
    conn = db.connect()
    try:
        result = conn.execute(sql)
        return result
    except Exception:
        print("Query failed")
        return None
    finally:
        conn.close()  # ALWAYS closes, even after return or exception

Raising Exceptions

raise — Throw Your Own Errors

# raise creates an exception manually — your code DECIDES something is wrong

def calculate_discount(price, discount_pct):
    if price < 0:
        raise ValueError(f"Price cannot be negative: {price}")
    if not 0 <= discount_pct <= 100:
        raise ValueError(f"Discount must be 0-100, got: {discount_pct}")
    return price * (1 - discount_pct / 100)

# Usage
try:
    final_price = calculate_discount(100, 150)
except ValueError as e:
    print(f"Invalid input: {e}")
    # Invalid input: Discount must be 0-100, got: 150

# Data engineering example: validate incoming data
def validate_record(record):
    required_fields = ["id", "name", "email"]
    missing = [f for f in required_fields if f not in record]
    if missing:
        raise ValueError(f"Missing required fields: {missing}")
    if not isinstance(record["id"], int):
        raise TypeError(f"ID must be int, got {type(record['id']).__name__}")
    if "@" not in record["email"]:
        raise ValueError(f"Invalid email: {record['email']}")

Re-raising Exceptions

# Catch an exception, do something (log it), then re-raise it
# Use bare "raise" (no arguments) to preserve the original traceback

def load_data(filepath):
    try:
        with open(filepath, "r") as f:
            return json.load(f)
    except json.JSONDecodeError as e:
        # Log the error for debugging
        print(f"CRITICAL: Corrupt JSON in {filepath}: {e}")
        log_to_monitoring(f"JSON parse failure: {filepath}")
        # Re-raise the SAME exception — let the caller decide what to do
        raise

# The caller sees the original exception with the full traceback
try:
    data = load_data("broken.json")
except json.JSONDecodeError:
    print("Using fallback data source")
    data = load_data("backup.json")

Raise with Custom Messages

# Wrap a low-level exception with a meaningful message
def connect_to_warehouse(config):
    try:
        conn = database.connect(
            host=config["host"],
            port=config["port"],
            database=config["database"]
        )
        return conn
    except ConnectionError as e:
        # Wrap with context — the caller sees YOUR message
        raise ConnectionError(
            f"Failed to connect to warehouse at {config['host']}:{config['port']}. "
            f"Original error: {e}"
        ) from e    # "from e" preserves the original exception chain

Custom Exceptions

Why Custom Exceptions

Built-in exceptions like ValueError are generic. When your pipeline raises a ValueError, is it a bad CSV row? An invalid config key? A data quality issue? Custom exceptions make errors self-documenting — the exception name tells you exactly what went wrong without reading the message.

# Generic — WHO raised this ValueError and WHY?
except ValueError:
    print("Something was invalid")

# Custom — immediately clear what happened
except DataQualityError:
    print("Data quality check failed — quarantine the batch")
except ConfigurationError:
    print("Pipeline config is invalid — check config.json")

Creating Custom Exceptions

# Custom exceptions are simple classes that inherit from Exception

class PipelineError(Exception):
    """Base exception for all pipeline errors."""
    pass

class DataQualityError(PipelineError):
    """Raised when data fails quality checks."""
    pass

class ConfigurationError(PipelineError):
    """Raised when pipeline configuration is invalid."""
    pass

class SourceConnectionError(PipelineError):
    """Raised when a source system is unreachable."""
    pass

# Usage
def validate_data(df):
    null_count = df["customer_id"].isnull().sum()
    if null_count > 0:
        raise DataQualityError(
            f"Found {null_count} null customer IDs — data cannot be loaded to gold layer"
        )

# Catching — you can catch the base class OR the specific subclass
try:
    validate_data(df)
    load_to_warehouse(df)
except DataQualityError as e:
    quarantine_batch(df, reason=str(e))
except PipelineError as e:
    # Catches ANY pipeline error (DataQuality, Configuration, SourceConnection)
    alert_team(f"Pipeline error: {e}")

Custom Exceptions with Extra Attributes

# Add context to your exceptions — table name, row count, error details

class DataQualityError(Exception):
    """Raised when data fails quality checks."""
    def __init__(self, message, table_name=None, failed_rows=0, check_name=None):
        super().__init__(message)
        self.table_name = table_name
        self.failed_rows = failed_rows
        self.check_name = check_name

# Raising with context
raise DataQualityError(
    "Null check failed on customer_id",
    table_name="bronze.customers",
    failed_rows=47,
    check_name="not_null"
)

# Catching with context
try:
    validate(df)
except DataQualityError as e:
    print(f"Quality check '{e.check_name}' failed on {e.table_name}")
    print(f"  {e.failed_rows} rows affected")
    print(f"  Details: {e}")
    # Quality check 'not_null' failed on bronze.customers
    #   47 rows affected
    #   Details: Null check failed on customer_id

Exception Groups (Data Pipeline Pattern)

# Collect ALL errors instead of stopping at the first one
# Useful for data validation — report everything wrong at once

class ValidationReport:
    def __init__(self):
        self.errors = []

    def add_error(self, check_name, message):
        self.errors.append({"check": check_name, "message": message})

    def has_errors(self):
        return len(self.errors) > 0

    def raise_if_errors(self):
        if self.errors:
            summary = "; ".join(f"{e['check']}: {e['message']}" for e in self.errors)
            raise DataQualityError(
                f"{len(self.errors)} validation errors: {summary}"
            )

# Usage — collect all errors, then raise once
report = ValidationReport()

if df["customer_id"].isnull().any():
    report.add_error("not_null", "customer_id has nulls")
if df["email"].str.contains("@").sum() < len(df):
    report.add_error("email_format", "invalid email addresses found")
if len(df) == 0:
    report.add_error("not_empty", "dataframe is empty")

report.raise_if_errors()
# DataQualityError: 2 validation errors: not_null: customer_id has nulls;
#                   email_format: invalid email addresses found

Nested try/except

# Try the primary approach, fall back to a secondary approach

def load_config():
    """Try JSON config, fall back to YAML, fall back to defaults."""
    # Try JSON first
    try:
        with open("config.json", "r") as f:
            return json.load(f)
    except FileNotFoundError:
        print("No JSON config — trying YAML")

    # Fall back to YAML
    try:
        import yaml
        with open("config.yaml", "r") as f:
            return yaml.safe_load(f)
    except (FileNotFoundError, ImportError):
        print("No YAML config — using defaults")

    # Final fallback
    return {"host": "localhost", "port": 5432, "database": "warehouse"}

# Primary → Secondary → Tertiary data source
def get_exchange_rate(currency):
    """Try live API, fall back to cache, fall back to default."""
    try:
        return fetch_from_api(currency)        # Live rate
    except ConnectionError:
        try:
            return read_from_cache(currency)   # Cached rate
        except FileNotFoundError:
            return 1.0                         # Default rate

Exception Chaining (from keyword)

# "from" preserves the ORIGINAL exception as the cause
# This creates a chain: your exception → original exception
# Critical for debugging — you see BOTH the high-level error and the root cause

class PipelineError(Exception):
    pass

def load_customer_data():
    try:
        with open("customers.csv", "r") as f:
            return f.read()
    except FileNotFoundError as e:
        raise PipelineError(
            "Customer data load failed — source file missing"
        ) from e

try:
    load_customer_data()
except PipelineError as e:
    print(f"Pipeline: {e}")
    print(f"Caused by: {e.__cause__}")
    # Pipeline: Customer data load failed — source file missing
    # Caused by: [Errno 2] No such file or directory: 'customers.csv'

# The traceback shows BOTH exceptions:
# FileNotFoundError: [Errno 2] No such file or directory: 'customers.csv'
#
# The above exception was the direct cause of the following exception:
#
# PipelineError: Customer data load failed — source file missing

Common Patterns for Data Engineers

Retry with Exponential Backoff

import time

def retry(func, max_attempts=3, base_delay=1):
    """Retry a function with exponential backoff."""
    for attempt in range(1, max_attempts + 1):
        try:
            return func()
        except (ConnectionError, TimeoutError) as e:
            if attempt == max_attempts:
                print(f"All {max_attempts} attempts failed. Last error: {e}")
                raise   # Re-raise on final attempt
            delay = base_delay * (2 ** (attempt - 1))   # 1s, 2s, 4s
            print(f"Attempt {attempt} failed: {e}. Retrying in {delay}s...")
            time.sleep(delay)

# Usage
result = retry(lambda: fetch_data_from_api("https://api.example.com/data"))

# Attempt 1 failed: Connection refused. Retrying in 1s...
# Attempt 2 failed: Connection refused. Retrying in 2s...
# (Attempt 3 succeeds or raises the exception)

Real-life analogy: Retry with backoff is like calling a busy restaurant. First call — busy, try again in 1 minute. Second call — still busy, try again in 2 minutes. Third call — still busy, try again in 4 minutes. You space out the calls so you are not hammering the phone line, giving the restaurant time to free up.

Skip Bad Records and Log

import csv

def process_csv_safely(filepath):
    """Process CSV rows, skip bad ones, log errors."""
    good_records = []
    bad_records = []

    with open(filepath, "r", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        for row_num, row in enumerate(reader, start=2):  # Start at 2 (header is row 1)
            try:
                # Validate and transform each row
                record = {
                    "id": int(row["id"]),
                    "name": row["name"].strip(),
                    "salary": float(row["salary"]),
                    "department": row["department"].strip(),
                }
                good_records.append(record)
            except (ValueError, KeyError) as e:
                bad_records.append({
                    "row_number": row_num,
                    "error": str(e),
                    "raw_data": dict(row),
                })

    print(f"Processed: {len(good_records)} good, {len(bad_records)} bad")

    # Write bad records to quarantine file
    if bad_records:
        with open("quarantine.json", "w") as f:
            json.dump(bad_records, f, indent=2)
        print(f"Bad records written to quarantine.json")

    return good_records

Validate Before Processing

from pathlib import Path

def validate_input(filepath, required_columns):
    """Validate a CSV file before processing."""
    path = Path(filepath)

    # Check file exists
    if not path.exists():
        raise FileNotFoundError(f"Input file not found: {filepath}")

    # Check file is not empty
    if path.stat().st_size == 0:
        raise ValueError(f"Input file is empty: {filepath}")

    # Check file extension
    if path.suffix.lower() not in (".csv", ".tsv"):
        raise ValueError(f"Expected CSV/TSV, got: {path.suffix}")

    # Check required columns exist
    with open(filepath, "r") as f:
        header = f.readline().strip().split(",")

    missing = [col for col in required_columns if col not in header]
    if missing:
        raise ValueError(f"Missing required columns: {missing}. Found: {header}")

    print(f"Validation passed: {filepath}")

# Usage — validate BEFORE any expensive processing
validate_input("customers.csv", ["id", "name", "email", "department"])

Database Connection Safety

# Always use try/finally or context managers for database connections

def execute_query(connection_string, query):
    """Execute a query with guaranteed connection cleanup."""
    conn = None
    try:
        conn = database.connect(connection_string)
        cursor = conn.cursor()
        cursor.execute(query)
        results = cursor.fetchall()
        conn.commit()
        return results
    except Exception as e:
        if conn:
            conn.rollback()    # Undo partial changes
        print(f"Query failed: {e}")
        raise
    finally:
        if conn:
            conn.close()       # ALWAYS close, even on error

# Better — use a context manager class (will cover in Decorators post)
# with DatabaseConnection(conn_string) as conn:
#     conn.execute(query)

API Error Handling

import requests
import time

def fetch_api_data(url, max_retries=3):
    """Fetch data from an API with retry and error handling."""
    for attempt in range(1, max_retries + 1):
        try:
            response = requests.get(url, timeout=30)
            response.raise_for_status()    # Raises HTTPError for 4xx/5xx
            return response.json()

        except requests.exceptions.Timeout:
            print(f"Attempt {attempt}: Request timed out")
        except requests.exceptions.ConnectionError:
            print(f"Attempt {attempt}: Connection failed")
        except requests.exceptions.HTTPError as e:
            if response.status_code == 429:    # Rate limited
                retry_after = int(response.headers.get("Retry-After", 60))
                print(f"Rate limited — waiting {retry_after}s")
                time.sleep(retry_after)
            elif response.status_code >= 500:  # Server error — retry
                print(f"Server error {response.status_code} — retrying")
            else:                               # Client error — do not retry
                print(f"Client error {response.status_code}: {e}")
                raise
        except requests.exceptions.JSONDecodeError:
            print(f"Response is not valid JSON")
            raise

        if attempt < max_retries:
            time.sleep(2 ** attempt)   # Exponential backoff

    raise ConnectionError(f"All {max_retries} attempts failed for {url}")

File Processing with Error Recovery

from pathlib import Path

def process_all_files(input_dir, output_dir):
    """Process all CSV files — continue even if some fail."""
    input_path = Path(input_dir)
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)

    results = {"success": [], "failed": []}

    for csv_file in sorted(input_path.glob("*.csv")):
        try:
            print(f"Processing: {csv_file.name}")
            data = load_and_transform(csv_file)
            save_output(data, output_path / csv_file.name)
            results["success"].append(csv_file.name)
        except Exception as e:
            print(f"  FAILED: {csv_file.name} — {e}")
            results["failed"].append({"file": csv_file.name, "error": str(e)})
            # Continue to next file — do NOT stop the pipeline

    print(f"
Results: {len(results['success'])} succeeded, {len(results['failed'])} failed")

    if results["failed"]:
        print("Failed files:")
        for f in results["failed"]:
            print(f"  {f['file']}: {f['error']}")

    return results

Anti-Patterns — What NOT to Do

# ❌ ANTI-PATTERN 1: Bare except — catches EVERYTHING including Ctrl+C
try:
    data = process()
except:                     # Catches KeyboardInterrupt, SystemExit — BAD
    print("Error occurred")
# ✅ FIX: Always catch Exception or a specific type
except Exception:
    print("Error occurred")

# ❌ ANTI-PATTERN 2: Catching and silencing — hides bugs
try:
    data = process()
except Exception:
    pass                    # Error is swallowed — you never know it happened
# ✅ FIX: At minimum, log the error
except Exception as e:
    print(f"Warning: {e}")  # Or log to file, monitoring, etc.

# ❌ ANTI-PATTERN 3: Catching too broadly — masks real bugs
try:
    config = load_config()
    data = fetch_data(config["url"])
    result = transform(data)
    save(result)
except Exception as e:      # Which step failed? No way to know!
    print(f"Pipeline failed: {e}")
# ✅ FIX: Wrap each step individually or use specific exceptions
try:
    config = load_config()
except FileNotFoundError:
    print("Config missing")

# ❌ ANTI-PATTERN 4: Using exceptions for flow control
try:
    value = my_dict[key]    # Expensive — exceptions are slow
except KeyError:
    value = default
# ✅ FIX: Use .get() or "in" check — faster and clearer
value = my_dict.get(key, default)

# ❌ ANTI-PATTERN 5: Catching Exception in a loop without limiting
while True:
    try:
        result = flaky_api()
        break
    except Exception:
        time.sleep(1)       # Infinite retry — never gives up, never alerts
# ✅ FIX: Always limit retries
for attempt in range(3):
    try:
        result = flaky_api()
        break
    except Exception:
        if attempt == 2: raise
        time.sleep(1)

Best Practices Summary

1. Catch specific exceptions — except ValueError not except Exception. Broad catches hide bugs.
2. Always use with for resources — files, connections, locks. No manual close() calls.
3. Never silence exceptions — except: pass is a bug factory. At minimum, log the error.
4. Use else for success code — keep it out of the try block so bugs crash loudly.
5. Use finally for cleanup — guaranteed to run even if the code crashes or returns early.
6. Raise early, catch late — validate inputs at function entry. Handle errors at the level that knows what to do about them.
7. Use custom exceptions for domains — DataQualityError is more informative than ValueError.
8. Use from for exception chaining — preserves the root cause for debugging.
9. Limit retries — always set a maximum. Infinite retries mask real problems.
10. Log before re-raising — capture context (file name, row number, config values) before propagating.

Common Mistakes

1. Bare except: without a type — catches KeyboardInterrupt and SystemExit, making the program impossible to stop. Always use except Exception at minimum.
2. except Exception as e: pass — silently swallows the error. The pipeline “succeeds” but produces wrong results. At minimum, log or print the exception.
3. Putting too much code in try — a broad try block masks which line actually failed. Keep try blocks small and focused on the one operation that might fail.
4. Catching exceptions for normal flow — using try/except KeyError instead of dict.get(). Exceptions are for exceptional situations, not routine lookups. They are also slower than conditional checks.
5. Not using from when re-raising — raise NewError("message") without from original loses the root cause. Always chain: raise NewError("message") from e.
6. Forgetting that finally runs after return — code in finally executes even if the try block returns. This is usually desirable (cleanup), but can be surprising if finally modifies a return value.

Interview Questions

Q: What is the difference between try/except and try/except/else/finally? A: try wraps the risky code. except handles specific exceptions. else runs only if no exception occurred — use it for success logic that should not be protected by except. finally always runs regardless of outcome — use it for cleanup like closing connections or writing logs.

Q: Why should you catch specific exceptions instead of using bare except? A: Bare except: catches everything including KeyboardInterrupt (Ctrl+C) and SystemExit, making the program impossible to stop gracefully. Specific exceptions let you handle each error type appropriately — a FileNotFoundError needs a different response than a ConnectionError. Broad catches also hide bugs by silently catching unexpected errors.

Q: What is the purpose of the finally clause? A: finally guarantees cleanup code runs regardless of what happens in the try block — whether it succeeds, an exception is caught, an unhandled exception propagates, or even if a return statement executes. It is essential for closing database connections, releasing file handles, and writing audit logs in data pipelines.

Q: What are custom exceptions and when should you use them? A: Custom exceptions are classes that inherit from Exception. Use them when built-in exceptions are too generic to describe domain-specific errors. For example, DataQualityError, ConfigurationError, and SourceConnectionError immediately tell you what failed without reading the message. They also let you catch groups of related errors via a common base class.

Q: What is exception chaining and why is it important? A: Exception chaining (using raise NewError() from original_error) preserves the root cause when you wrap a low-level exception in a higher-level one. The traceback shows both exceptions, making debugging much easier. Without from, the original error is lost, and you only see the wrapper exception.

Q: How would you handle errors in a data pipeline that processes multiple files? A: Wrap each file’s processing in its own try/except block inside the loop. On failure, log the error with the filename and continue to the next file. Collect results (success/fail counts and error details) and report at the end. This prevents one bad file from stopping the entire pipeline.

Wrapping Up

Error handling is the difference between a script and a production system. The try/except/else/finally structure gives you precise control over what happens when things go wrong. Catch specific exceptions. Use else for success logic. Use finally for guaranteed cleanup. Raise custom exceptions that describe your domain. And always log errors with enough context to debug them at 3 AM.

In data engineering, the patterns matter most: retry with backoff for flaky connections, skip-and-quarantine for bad records, validate-before-processing for early failure, and continue-on-error for batch file processing. These patterns keep your pipelines resilient without hiding real problems.

Previous in this series: File Handling — Read, Write, CSV, JSON, Context Managers

Next in this series: Logging — Levels, Formatters, Handlers, and Production Patterns

---

Naveen Vuppula is a Senior Data Engineering Consultant and app developer based in Ontario, Canada. He writes about Python, SQL, AWS, Azure, and everything data engineering at DriveDataScience.com.