Custom Exceptions - Designing Your Exception Hierarchy
Reading time: ~18 minutes | Level: Foundation → Engineering
Here is something most Python developers do not think to try:
class InsufficientFundsError(Exception):
pass
def withdraw(balance, amount):
if amount > balance:
raise InsufficientFundsError(f"Cannot withdraw {amount}, balance is {balance}")
return balance - amount
try:
withdraw(100, 250)
except ValueError:
print("Caught ValueError")
except InsufficientFundsError as e:
print(f"Caught domain error: {e}")
Caught domain error: Cannot withdraw 250, balance is 100
The except ValueError block does not fire. The except InsufficientFundsError block does. That is the entire point of custom exceptions: they let you catch exactly your errors and nothing else.
But there is far more to custom exceptions than that single line of inheritance. This page covers how to design exception hierarchies that carry structured data, wrap third-party errors, and communicate domain semantics with the same precision a well-designed API does.
What You Will Learn
- Why custom exceptions exist and when to create them instead of reusing built-ins
- The difference between inheriting from
ExceptionversusBaseException - How to add rich context: custom
__init__, extra attributes, machine-readable data - How to write
__str__and__repr__for readable error messages - How to design a multi-level exception hierarchy for a library or domain
- How to wrap third-party exceptions using
raise MyError(...) from original - Mixin exceptions and why they enable powerful
isinstancechecks - Exception naming conventions from PEP 8
- Real-world patterns from FastAPI, SQLAlchemy, and payment systems
Prerequisites
- Python try/except/finally (topic 02 in this module)
- Python exception hierarchy and built-in exceptions (topic 03)
- Raising exceptions with
raise(topic 04) - Python class basics - defining a class,
__init__, inheritance
The Problem Custom Exceptions Solve
Built-in exceptions are general-purpose. They describe what went wrong mechanically, not why it matters to your domain:
# Bad: using a generic exception
def withdraw(balance, amount, account_id):
if amount > balance:
raise ValueError(f"Cannot withdraw {amount}")
Three problems with this:
- Callers cannot distinguish your error from every other
ValueErrorin the call stack - a typo, a bad conversion, a mismatch in configuration. - No structured data: the caller gets a string. To extract the amount or balance programmatically they have to parse the message - fragile.
- No domain semantics:
ValueErrorsays nothing about accounts, funds, or banking rules.
Compare to:
class InsufficientFundsError(Exception):
def __init__(self, account_id, amount, balance):
self.account_id = account_id
self.amount = amount
self.balance = balance
super().__init__(
f"Account {account_id}: cannot withdraw {amount:.2f}, "
f"available balance is {balance:.2f}"
)
Now the caller can:
- Catch only
InsufficientFundsError, not everyValueError - Read
e.account_id,e.amount,e.balanceas structured data - Log or display a precise human message via
str(e)
Part 1 - Inheriting from Exception, Not BaseException
Always inherit from Exception (or a subclass of it), never from BaseException directly.
Why? Because except Exception is the standard broad catch in Python. If you inherit from BaseException, your exception escapes these guards:
# This does NOT catch BaseException subclasses
try:
some_operation()
except Exception as e:
log_error(e) # Your BaseException subclass would slip through!
SystemExit, KeyboardInterrupt, and GeneratorExit inherit from BaseException precisely so they cannot be silently swallowed by except Exception. Your domain errors should not have that behavior.
:::danger Never Inherit from BaseException
Inheriting your custom exceptions from BaseException is a serious design error. It means except Exception will not catch your errors, making them nearly impossible to handle gracefully in frameworks and libraries that use broad catches for logging and cleanup.
:::
Part 2 - The Minimal Custom Exception
The simplest valid custom exception:
class DatabaseConnectionError(Exception):
pass
That is it. One line of inheritance. It gives you:
- A distinct type that callers can catch specifically
- The full
Exceptioninterface: message,args,__str__,__repr__ - Membership in the normal exception hierarchy
raise DatabaseConnectionError("Failed to connect to postgres://localhost:5432")
# Callers can do:
try:
connect()
except DatabaseConnectionError as e:
print(f"DB error: {e}") # DB error: Failed to connect to postgres://localhost:5432
reconnect()
except Exception as e:
print(f"Unexpected: {e}")
raise
When is the minimal form enough? When the message alone carries all the context a caller needs and when no programmatic inspection of the error fields is required.
Part 3 - Adding Rich Context with Custom __init__
Most real-world custom exceptions need structured data, not just a message string.
The Pattern
class InsufficientFundsError(Exception):
"""Raised when a withdrawal exceeds the available account balance."""
def __init__(self, account_id: str, amount: float, balance: float):
self.account_id = account_id
self.amount = amount
self.balance = balance
self.shortfall = amount - balance
# Always call super().__init__() with the human-readable message
super().__init__(
f"Account '{account_id}': cannot withdraw {amount:.2f}, "
f"balance is {balance:.2f} (shortfall: {self.shortfall:.2f})"
)
# Raising it
try:
raise InsufficientFundsError("ACC-001", amount=500.00, balance=320.75)
except InsufficientFundsError as e:
print(str(e))
# Account 'ACC-001': cannot withdraw 500.00, balance is 320.75 (shortfall: 179.25)
# Structured data is available as attributes
print(f"Account: {e.account_id}") # Account: ACC-001
print(f"Shortfall: {e.shortfall}") # Shortfall: 179.25
# A payment system could use this to auto-top-up:
if e.shortfall < 50.00:
auto_top_up(e.account_id, e.shortfall)
The key discipline: call super().__init__(human_message). This ensures str(e) and repr(e) and logging all work correctly.
Custom __str__ and __repr__
If you want full control over the string representations:
class APIError(Exception):
def __init__(self, status_code: int, endpoint: str, detail: str):
self.status_code = status_code
self.endpoint = endpoint
self.detail = detail
super().__init__(detail) # minimal args to super
def __str__(self):
return f"HTTP {self.status_code} at {self.endpoint}: {self.detail}"
def __repr__(self):
return (
f"APIError(status_code={self.status_code!r}, "
f"endpoint={self.endpoint!r}, "
f"detail={self.detail!r})"
)
err = APIError(404, "/api/users/99", "User not found")
print(str(err)) # HTTP 404 at /api/users/99: User not found
print(repr(err)) # APIError(status_code=404, endpoint='/api/users/99', detail='User not found')
# In logs, repr() is used when the object is inside a container:
errors = [err]
print(errors)
# [APIError(status_code=404, endpoint='/api/users/99', detail='User not found')]
:::tip When to Define str vs repr
Define __repr__ when you want a developer-readable, unambiguous representation (useful in logs and debug sessions). Define __str__ when the human message differs significantly from the technical representation. If you only define one, prefer __repr__ - Python falls back to it for str() when __str__ is missing.
:::
Part 4 - Designing a Library Exception Hierarchy
Professional libraries (requests, SQLAlchemy, Pydantic, FastAPI) all define a hierarchy rooted in a single base exception. This lets users write either a broad catch or a specific catch:
# Broad catch: catch any error from your library
except MyLibraryError:
...
# Specific catch: handle only one scenario
except MyLibraryConnectionError:
...
Example: A Data Pipeline Library
# errors.py - the entire exception module for a data pipeline library
class PipelineError(Exception):
"""Base exception for all pipeline library errors.
All exceptions raised by this library inherit from PipelineError.
Users can catch PipelineError to handle any library error, or catch
specific subclasses for targeted handling.
"""
pass
# ── Source errors ─────────────────────────────────────────────────────────────
class SourceError(PipelineError):
"""Errors reading from a data source."""
pass
class SourceConnectionError(SourceError):
"""Cannot connect to the data source."""
def __init__(self, source_url: str, reason: str):
self.source_url = source_url
self.reason = reason
super().__init__(f"Cannot connect to {source_url!r}: {reason}")
class SourceTimeoutError(SourceError):
"""Connection to source timed out."""
def __init__(self, source_url: str, timeout_seconds: float):
self.source_url = source_url
self.timeout_seconds = timeout_seconds
super().__init__(
f"Source {source_url!r} timed out after {timeout_seconds}s"
)
# ── Transform errors ──────────────────────────────────────────────────────────
class TransformError(PipelineError):
"""Errors during data transformation."""
pass
class SchemaValidationError(TransformError):
"""Input data does not match expected schema."""
def __init__(self, field: str, expected_type: type, got_value):
self.field = field
self.expected_type = expected_type
self.got_value = got_value
super().__init__(
f"Field '{field}': expected {expected_type.__name__}, "
f"got {type(got_value).__name__} = {got_value!r}"
)
class DataRangeError(TransformError):
"""A value is outside the expected range."""
def __init__(self, field: str, value, min_val=None, max_val=None):
self.field = field
self.value = value
self.min_val = min_val
self.max_val = max_val
bounds = f"[{min_val}, {max_val}]"
super().__init__(f"Field '{field}' value {value!r} out of range {bounds}")
# ── Sink errors ───────────────────────────────────────────────────────────────
class SinkError(PipelineError):
"""Errors writing to a data sink."""
pass
class SinkWriteError(SinkError):
"""Failed to write a batch of records."""
def __init__(self, sink_name: str, batch_size: int, reason: str):
self.sink_name = sink_name
self.batch_size = batch_size
self.reason = reason
super().__init__(
f"Failed to write {batch_size} records to '{sink_name}': {reason}"
)
The hierarchy visualised:
Using this hierarchy:
from pipeline.errors import (
PipelineError, SourceTimeoutError, SchemaValidationError
)
def run_pipeline(config):
try:
data = source.read()
except SourceTimeoutError as e:
# Retry logic - we know exactly what went wrong
print(f"Retrying after timeout ({e.timeout_seconds}s): {e.source_url}")
data = source.read(timeout=e.timeout_seconds * 2)
except SourceError as e:
# Other source errors - can't recover
raise PipelineError(f"Pipeline aborted: {e}") from e
try:
transformed = transform(data)
except SchemaValidationError as e:
print(f"Bad data in field '{e.field}': got {type(e.got_value).__name__}")
raise
except PipelineError:
raise # Re-raise any other pipeline error unchanged
Part 5 - Exception Chaining: raise ... from
When you catch a third-party exception and raise your own, you should always preserve the original exception as context. Python provides raise NewError(...) from original_exc for this.
import sqlite3
class DatabaseError(Exception):
"""Base error for our database layer."""
pass
class RecordNotFoundError(DatabaseError):
def __init__(self, table: str, record_id):
self.table = table
self.record_id = record_id
super().__init__(f"No record with id={record_id!r} in table '{table}'")
class DuplicateRecordError(DatabaseError):
def __init__(self, table: str, constraint: str):
self.table = table
self.constraint = constraint
super().__init__(f"Duplicate record in '{table}' (constraint: {constraint})")
def get_user(db, user_id: int):
try:
row = db.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
if row is None:
raise RecordNotFoundError("users", user_id)
return dict(row)
except sqlite3.OperationalError as e:
# Wrap the DB driver error in our domain error
raise DatabaseError(f"Query failed for user {user_id}") from e
def create_user(db, email: str, name: str):
try:
db.execute(
"INSERT INTO users (email, name) VALUES (?, ?)", (email, name)
)
db.commit()
except sqlite3.IntegrityError as e:
# sqlite3.IntegrityError is wrapped into our domain error
raise DuplicateRecordError("users", "users_email_unique") from e
When chained, Python displays both:
Traceback (most recent call last):
File "db.py", line 8, in create_user
db.execute("INSERT ...")
sqlite3.IntegrityError: UNIQUE constraint failed: users.email
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "main.py", line 15, in <module>
create_user(db, "[email protected]", "Alice")
DuplicateRecordError: Duplicate record in 'users' (constraint: users_email_unique)
The original exception is also available programmatically:
try:
except DuplicateRecordError as e:
print(e.__cause__) # The original sqlite3.IntegrityError
print(e.__context__) # Same when using 'raise from'
:::note Suppress the Original Exception with from None
If you deliberately do not want the original exception shown (for security - hiding internal DB details), use raise MyError(...) from None. The __cause__ will be None and Python will not display the original traceback.
:::
# Hide internal implementation details from callers
raise DatabaseError("A database error occurred") from None
Part 6 - Mixin Exceptions
Sometimes an exception genuinely belongs to two categories. Python allows multiple inheritance for exceptions, enabling mixin patterns:
class TimeoutError(Exception):
"""A generic timeout."""
pass
class NetworkError(Exception):
"""A generic network error."""
pass
class NetworkTimeoutError(TimeoutError, NetworkError):
"""Timed out waiting for a network response.
Inherits from both TimeoutError and NetworkError so callers can
catch either the general category or the specific combined type.
"""
def __init__(self, url: str, timeout_seconds: float):
self.url = url
self.timeout_seconds = timeout_seconds
super().__init__(f"Network timeout after {timeout_seconds}s: {url}")
err = NetworkTimeoutError("https://api.example.com/data", 30.0)
print(isinstance(err, NetworkTimeoutError)) # True
print(isinstance(err, TimeoutError)) # True
print(isinstance(err, NetworkError)) # True
print(isinstance(err, Exception)) # True
# All three except clauses would catch this:
try:
raise err
except NetworkTimeoutError:
print("Caught as NetworkTimeoutError") # This fires first (most specific)
This is the same pattern Python uses for FileNotFoundError, which is both an OSError and has a subtype relationship with IOError.
:::warning MRO Complexity with Mixin Exceptions Multiple inheritance with exceptions works fine in practice but can produce surprising MRO (Method Resolution Order) behaviour if the hierarchy is deep. Keep mixin exception hierarchies shallow - two levels of multiple inheritance is usually the maximum you will need. :::
Part 7 - When to Create a New Exception vs Reuse Built-ins
The rule of thumb:
| Decision | Create a New Exception When... |
|---|---|
| Domain specificity | The error is domain-specific (e.g., "payment declined") - not meaningfully expressed by any built-in |
| Targeted catching | Callers need to catch your errors independently of others in the same category |
| Structured data | You need to attach structured data (amounts, IDs, URLs) |
| Library API | You are building a library and need a stable public API for error handling |
| Decision | Reuse a Built-in When... |
|---|---|
| General error | The error is truly general: wrong type → TypeError, bad value → ValueError, bad key → KeyError |
| No distinction needed | Callers do not need to distinguish your error from others of the same category |
| Small scope | You are writing a short script or internal utility |
The grey zone: subclass a built-in when the error is a specialised form of a known category: class NegativeAmountError(ValueError): pass
Subclassing Built-ins for Specialisation
class NegativeAmountError(ValueError):
"""A monetary amount that must be positive is negative or zero."""
def __init__(self, amount: float, field: str = "amount"):
self.amount = amount
self.field = field
super().__init__(f"'{field}' must be positive, got {amount}")
class EmptyBatchError(ValueError):
"""A batch operation received an empty collection."""
def __init__(self, operation: str):
self.operation = operation
super().__init__(f"Cannot run '{operation}' on an empty batch")
Now callers can catch ValueError (broad) or NegativeAmountError (specific). This is the same relationship FileNotFoundError has to OSError.
Part 8 - PEP 8 Naming Conventions
PEP 8 is explicit:
Exception classes should end in "Error" if the exception is an error.
# Correct
class InsufficientFundsError(Exception): pass
class ConnectionTimeoutError(Exception): pass
class SchemaValidationError(Exception): pass
# Wrong - missing "Error" suffix
class InsufficientFunds(Exception): pass # Bad
class InvalidConfig(Exception): pass # Bad
# Exception: non-error exceptions (used for control flow) may omit "Error"
class StopIteration(Exception): pass # built-in, control flow
class EndOfStream(Exception): pass # acceptable for control flow exceptions
:::note Base Exception Naming
The module-level base exception for your library often simply ends in Error: PipelineError, PaymentError, DatabaseError. Specific subclasses get more descriptive names: PipelineSchemaError, PaymentDeclinedError.
:::
Part 9 - Exception Attributes as Machine-Readable Data
Well-designed exceptions carry attributes that code - not just humans - can consume:
class PaymentDeclinedError(Exception):
"""A payment processor declined a charge."""
# Standardised decline codes from the payment processor
INSUFFICIENT_FUNDS = "insufficient_funds"
CARD_EXPIRED = "card_expired"
FRAUD_SUSPECTED = "fraud_suspected"
CARD_BLOCKED = "card_blocked"
def __init__(
self,
decline_code: str,
amount: float,
currency: str,
processor_message: str,
retry_allowed: bool = False,
):
self.decline_code = decline_code
self.amount = amount
self.currency = currency
self.processor_message = processor_message
self.retry_allowed = retry_allowed
super().__init__(
f"Payment declined [{decline_code}]: {processor_message} "
f"(amount: {amount:.2f} {currency})"
)
def to_dict(self) -> dict:
"""Return a JSON-serialisable representation for API responses."""
return {
"error": "payment_declined",
"decline_code": self.decline_code,
"amount": self.amount,
"currency": self.currency,
"message": self.processor_message,
"retry_allowed": self.retry_allowed,
}
try:
charge(card_token="tok_xyz", amount=99.99, currency="USD")
except PaymentDeclinedError as e:
if e.retry_allowed:
# Programmatic decision based on structured attribute
return {"status": "retry", **e.to_dict()}
elif e.decline_code == PaymentDeclinedError.FRAUD_SUSPECTED:
alert_fraud_team(e)
return {"status": "blocked", **e.to_dict()}
else:
return {"status": "failed", **e.to_dict()}
The caller does not parse the human message. It reads structured attributes. This is the key design discipline.
Part 10 - Real-World Patterns
FastAPI: HTTPException
FastAPI defines its own HTTPException that carries HTTP-specific attributes:
# Simplified version of FastAPI's HTTPException
class HTTPException(Exception):
def __init__(self, status_code: int, detail: str = None, headers: dict = None):
self.status_code = status_code
self.detail = detail
self.headers = headers
super().__init__(detail)
# Usage in a route handler
from fastapi import HTTPException
@app.get("/users/{user_id}")
def get_user(user_id: int, db=Depends(get_db)):
user = db.query(User).filter(User.id == user_id).first()
if user is None:
raise HTTPException(status_code=404, detail="User not found")
return user
FastAPI's global exception handler catches HTTPException and converts it to an HTTP response using e.status_code and e.detail. No string parsing required.
SQLAlchemy: IntegrityError Wrapping
A common FastAPI + SQLAlchemy pattern wraps the ORM's constraint errors:
from sqlalchemy.exc import IntegrityError
from fastapi import HTTPException
@app.post("/users/")
def create_user(user: UserCreate, db: Session = Depends(get_db)):
db_user = User(email=user.email, name=user.name)
db.add(db_user)
try:
db.commit()
except IntegrityError as e:
db.rollback()
# Wrap SQLAlchemy's internal error into a clean HTTP error
raise HTTPException(
status_code=409,
detail=f"A user with email '{user.email}' already exists"
) from e
db.refresh(db_user)
return db_user
The from e preserves the original IntegrityError in the traceback for debugging while exposing only the clean HTTPException to the API layer.
Domain Exception Hierarchies in Payment Systems
# A realistic payment domain exception hierarchy
class PaymentError(Exception):
"""Root of all payment-domain errors."""
pass
class PaymentValidationError(PaymentError, ValueError):
"""Input data is invalid - inherits from ValueError for broad catches."""
pass
class PaymentProcessingError(PaymentError):
"""An error occurred during payment processing."""
def __init__(self, transaction_id: str, reason: str, retryable: bool = False):
self.transaction_id = transaction_id
self.reason = reason
self.retryable = retryable
super().__init__(f"Transaction {transaction_id} failed: {reason}")
class PaymentDeclinedError(PaymentProcessingError):
"""The payment processor declined the charge."""
pass
class PaymentNetworkError(PaymentProcessingError):
"""Network failure during payment processing - usually retryable."""
def __init__(self, transaction_id: str, reason: str):
super().__init__(transaction_id, reason, retryable=True)
class PaymentFraudError(PaymentProcessingError):
"""Transaction flagged as potentially fraudulent."""
def __init__(self, transaction_id: str, risk_score: float):
self.risk_score = risk_score
super().__init__(
transaction_id,
f"Fraud risk score {risk_score:.2f} exceeds threshold",
retryable=False
)
Interview Questions
Q1: Why should custom exceptions inherit from Exception rather than BaseException?
Answer: BaseException is the root of everything including SystemExit, KeyboardInterrupt, and GeneratorExit. These three exist specifically so they cannot be caught by except Exception - they represent signals that need to propagate up (e.g., Ctrl+C should terminate the program). If your domain exception inherits from BaseException, it also bypasses except Exception handlers used by frameworks for logging, cleanup, and HTTP error conversion. The result is your error propagating in unexpected ways through library code. Always inherit from Exception so your errors participate normally in the exception handling ecosystem.
Q2: What is exception chaining (raise X from Y) and when should you use it?
Answer: Exception chaining links a new exception to the original that caused it. raise NewError("...") from original_exc sets NewError.__cause__ = original_exc and Python displays both in the traceback with "The above exception was the direct cause of the following exception." Use it when translating low-level or third-party exceptions into your domain exceptions - for example wrapping sqlite3.IntegrityError into DuplicateRecordError. This preserves the original context for debugging while exposing a clean API. Use from None to suppress the original entirely (useful when hiding implementation details from API callers for security reasons).
Q3: What is the purpose of calling super().__init__(message) in a custom exception?
Answer: The Exception base class stores its arguments in self.args and uses them to implement __str__ and __repr__. If you do not call super().__init__(), then str(e) returns an empty string, logging produces no useful output, and standard tools that display exceptions (debuggers, Sentry, log aggregators) show nothing informative. Always pass the human-readable message as the first argument to super().__init__(). Your custom attributes can be set before or after the super call - order does not matter.
Q4: When should you subclass a built-in exception (like ValueError) versus creating a new root exception?
Answer: Subclass a built-in when your error IS a specialised form of a known general category. NegativeAmountError(ValueError) makes sense because a negative amount is a bad value. EmptyBatchError(ValueError) makes sense because an empty batch is an invalid argument. Callers who write except ValueError will catch these automatically, which is correct behaviour. Create a new root exception (inheriting from Exception directly) when your error does not meaningfully fit a built-in category, or when you need a base type for a family of domain-specific errors that callers must explicitly opt in to handle - for example PaymentError as the root of a payment library's hierarchy.
Q5: What is the mixin exception pattern and what problem does it solve?
Answer: Mixin exceptions use multiple inheritance so a single exception belongs to two or more exception categories simultaneously. For example NetworkTimeoutError(TimeoutError, NetworkError) is both a timeout and a network error. This lets callers catch it by the most specific type (NetworkTimeoutError), by the timeout category (TimeoutError), or by the network category (NetworkError) - all with simple isinstance or except logic. The problem it solves: an error is genuinely a member of multiple independent categories, and restricting it to one would force callers to either over-catch (too broad) or under-catch (miss the error in one category). Python uses this pattern itself - FileNotFoundError inherits from both OSError and can be caught as OSError, IOError, or FileNotFoundError.
Q6: How do you design exception attributes to be useful for programmatic error handling rather than just human reading?
Answer: The key discipline is: store structured data as attributes, not embedded in the message string. The message is for humans. The attributes are for code. For example, a PaymentDeclinedError should have e.decline_code, e.amount, e.currency, and e.retry_allowed as separate attributes. Callers can then branch on e.decline_code == "insufficient_funds" without parsing the string. Additionally, add a to_dict() method that returns a JSON-serialisable representation for API responses, logging, and monitoring. This makes your exception a data carrier, not just a signal.
Practice Challenges
Beginner - Minimal Custom Exception
Problem: A function parse_age(value) should raise a custom InvalidAgeError if the value is not a positive integer less than 150. The error message should include the invalid value and why it is invalid.
Solution
class InvalidAgeError(ValueError):
"""Raised when an age value fails validation."""
def __init__(self, value, reason: str):
self.value = value
self.reason = reason
super().__init__(f"Invalid age {value!r}: {reason}")
def parse_age(value) -> int:
"""Parse and validate an age value.
Args:
value: The raw value to parse (may be any type).
Returns:
The validated age as an integer.
Raises:
InvalidAgeError: If the value is not a valid age.
"""
if not isinstance(value, int):
raise InvalidAgeError(value, f"expected int, got {type(value).__name__}")
if value < 0:
raise InvalidAgeError(value, "age cannot be negative")
if value >= 150:
raise InvalidAgeError(value, "age must be less than 150")
return value
# Test it
test_cases = [25, -1, 200, "thirty", 0, 149]
for case in test_cases:
try:
age = parse_age(case)
print(f"OK: age = {age}")
except InvalidAgeError as e:
print(f"INVALID: {e}")
print(f" value={e.value!r}, reason={e.reason!r}")
# Output:
# OK: age = 25
# INVALID: Invalid age -1: age cannot be negative
# value=-1, reason='age cannot be negative'
# INVALID: Invalid age 200: age must be less than 150
# value=200, reason='age must be less than 150'
# INVALID: Invalid age 'thirty': expected int, got str
# value='thirty', reason='expected int, got str'
# OK: age = 0
# OK: age = 149
Intermediate - Full Exception Hierarchy
Problem: Design an exception hierarchy for a file-processing library. The library reads CSV files, validates their schema, and writes processed output. Define at least five exceptions with appropriate attributes and a proper hierarchy. Demonstrate their use in a realistic function.
Solution
# ── Exception hierarchy ───────────────────────────────────────────────────────
class FileProcessorError(Exception):
"""Root exception for the file-processing library."""
pass
class FileReadError(FileProcessorError):
"""Cannot read the input file."""
def __init__(self, path: str, reason: str):
self.path = path
self.reason = reason
super().__init__(f"Cannot read '{path}': {reason}")
class CSVParseError(FileProcessorError):
"""The file is not valid CSV."""
def __init__(self, path: str, line_number: int, raw_line: str):
self.path = path
self.line_number = line_number
self.raw_line = raw_line
super().__init__(
f"CSV parse error in '{path}' at line {line_number}: {raw_line!r}"
)
class ColumnMissingError(FileProcessorError):
"""A required column is absent from the CSV header."""
def __init__(self, path: str, column: str, available_columns: list):
self.path = path
self.column = column
self.available_columns = available_columns
super().__init__(
f"Required column '{column}' missing in '{path}'. "
f"Available: {available_columns}"
)
class ValueValidationError(FileProcessorError):
"""A cell value fails validation rules."""
def __init__(self, path: str, row: int, column: str, value, reason: str):
self.path = path
self.row = row
self.column = column
self.value = value
self.reason = reason
super().__init__(
f"Validation failed at '{path}' row {row}, column '{column}': "
f"{value!r} - {reason}"
)
class FileWriteError(FileProcessorError):
"""Cannot write the output file."""
def __init__(self, path: str, reason: str):
self.path = path
self.reason = reason
super().__init__(f"Cannot write '{path}': {reason}")
# ── Library function using the hierarchy ──────────────────────────────────────
import csv
import io
REQUIRED_COLUMNS = {"name", "age", "email"}
def process_csv(input_path: str, output_path: str) -> int:
"""Read, validate, and write a CSV file. Returns row count processed."""
# Step 1: Read
try:
with open(input_path, "r", newline="") as f:
content = f.read()
except OSError as e:
raise FileReadError(input_path, str(e)) from e
# Step 2: Parse
try:
reader = csv.DictReader(io.StringIO(content))
rows = list(reader)
except csv.Error as e:
raise CSVParseError(input_path, 0, str(e)) from e
# Step 3: Validate schema
if reader.fieldnames is None:
raise ColumnMissingError(input_path, "any column", [])
actual_columns = set(reader.fieldnames)
for col in REQUIRED_COLUMNS:
if col not in actual_columns:
raise ColumnMissingError(input_path, col, list(actual_columns))
# Step 4: Validate values
processed = []
for row_num, row in enumerate(rows, start=2):
# Validate age
try:
age = int(row["age"])
if age < 0 or age > 120:
raise ValueValidationError(
input_path, row_num, "age", row["age"],
"must be between 0 and 120"
)
except ValueError:
raise ValueValidationError(
input_path, row_num, "age", row["age"], "must be an integer"
)
# Validate email (minimal check)
if "@" not in row["email"]:
raise ValueValidationError(
input_path, row_num, "email", row["email"],
"must contain '@'"
)
processed.append({**row, "age": age})
# Step 5: Write output
try:
with open(output_path, "w", newline="") as f:
writer = csv.DictWriter(f, fieldnames=list(REQUIRED_COLUMNS))
writer.writeheader()
writer.writerows(processed)
except OSError as e:
raise FileWriteError(output_path, str(e)) from e
return len(processed)
# ── Caller handling each error category ──────────────────────────────────────
def safe_process(input_path: str, output_path: str):
try:
count = process_csv(input_path, output_path)
print(f"Processed {count} rows successfully")
except ColumnMissingError as e:
print(f"Schema error: column '{e.column}' not found")
print(f"Available columns: {e.available_columns}")
except ValueValidationError as e:
print(f"Data error at row {e.row}, column '{e.column}':")
print(f" Value: {e.value!r}")
print(f" Reason: {e.reason}")
except FileReadError as e:
print(f"Cannot read file: {e.path}")
except FileWriteError as e:
print(f"Cannot write file: {e.path}")
except FileProcessorError as e:
# Catch-all for any other library error
print(f"Processing error: {e}")
Advanced - Exception Hierarchy with Exception Chaining and to_dict()
Problem: Build a minimal HTTP client exception hierarchy. The hierarchy must: (1) wrap urllib.error exceptions into domain exceptions with exception chaining, (2) carry structured attributes for programmatic handling, (3) expose a to_dict() method for JSON serialisation, and (4) include a retry decision via a retryable property.
Solution
import urllib.error
import urllib.request
import json
from typing import Optional
# ── Exception hierarchy ───────────────────────────────────────────────────────
class HTTPClientError(Exception):
"""Root exception for the HTTP client library."""
@property
def retryable(self) -> bool:
return False
def to_dict(self) -> dict:
return {"error": type(self).__name__, "message": str(self)}
class HTTPRequestError(HTTPClientError):
"""A request could not be sent (network-level error)."""
def __init__(self, url: str, reason: str):
self.url = url
self.reason = reason
super().__init__(f"Request to {url!r} failed: {reason}")
@property
def retryable(self) -> bool:
return True # Network errors are usually transient
def to_dict(self) -> dict:
return {
"error": "request_error",
"url": self.url,
"reason": self.reason,
"retryable": self.retryable,
}
class HTTPResponseError(HTTPClientError):
"""The server returned an error status code."""
def __init__(self, url: str, status_code: int, body: Optional[str] = None):
self.url = url
self.status_code = status_code
self.body = body
super().__init__(f"HTTP {status_code} from {url!r}")
@property
def retryable(self) -> bool:
# 429 Too Many Requests and 5xx server errors are retryable
return self.status_code == 429 or self.status_code >= 500
def to_dict(self) -> dict:
return {
"error": "response_error",
"url": self.url,
"status_code": self.status_code,
"body": self.body,
"retryable": self.retryable,
}
class HTTPTimeoutError(HTTPRequestError):
"""The request timed out."""
def __init__(self, url: str, timeout_seconds: float):
self.timeout_seconds = timeout_seconds
super().__init__(url, f"timed out after {timeout_seconds}s")
def to_dict(self) -> dict:
return {
**super().to_dict(),
"error": "timeout_error",
"timeout_seconds": self.timeout_seconds,
}
class HTTPAuthError(HTTPResponseError):
"""The server rejected the request due to authentication failure."""
def __init__(self, url: str):
super().__init__(url, status_code=401)
@property
def retryable(self) -> bool:
return False # Auth errors are not transient
def to_dict(self) -> dict:
return {**super().to_dict(), "error": "auth_error"}
# ── Client function wrapping urllib ──────────────────────────────────────────
def fetch_json(url: str, timeout: float = 10.0) -> dict:
"""Fetch JSON from a URL.
Raises:
HTTPTimeoutError: Request timed out.
HTTPAuthError: Server returned 401.
HTTPResponseError: Server returned any other error status.
HTTPRequestError: Network-level failure.
"""
try:
with urllib.request.urlopen(url, timeout=timeout) as response:
body = response.read().decode("utf-8")
return json.loads(body)
except TimeoutError as e:
raise HTTPTimeoutError(url, timeout) from e
except urllib.error.HTTPError as e:
body = e.read().decode("utf-8") if e.fp else None
if e.code == 401:
raise HTTPAuthError(url) from e
raise HTTPResponseError(url, e.code, body) from e
except urllib.error.URLError as e:
raise HTTPRequestError(url, str(e.reason)) from e
# ── Caller with retry logic ──────────────────────────────────────────────────
import time
def fetch_with_retry(url: str, max_retries: int = 3, backoff: float = 1.0) -> dict:
last_error = None
for attempt in range(1, max_retries + 1):
try:
return fetch_json(url)
except HTTPClientError as e:
last_error = e
if not e.retryable:
print(f"Non-retryable error: {e}")
print(f"Details: {json.dumps(e.to_dict(), indent=2)}")
raise
wait = backoff * (2 ** (attempt - 1))
print(f"Attempt {attempt} failed ({e}). Retrying in {wait}s...")
time.sleep(wait)
print(f"All {max_retries} attempts failed.")
raise last_error
# ── Demonstrate the hierarchy ─────────────────────────────────────────────────
errors = [
HTTPTimeoutError("https://api.example.com/data", 30.0),
HTTPAuthError("https://api.example.com/secure"),
HTTPResponseError("https://api.example.com/data", 500, "Internal Server Error"),
HTTPRequestError("https://api.example.com/data", "Name or service not known"),
]
for err in errors:
print(f"Error: {err}")
print(f" retryable: {err.retryable}")
print(f" to_dict: {err.to_dict()}")
print(f" isinstance HTTPClientError: {isinstance(err, HTTPClientError)}")
print()
# Output (formatted):
# Error: Request to 'https://api.example.com/data' failed: timed out after 30.0s
# retryable: True
# to_dict: {'error': 'timeout_error', 'url': ..., 'reason': ..., 'retryable': True, 'timeout_seconds': 30.0}
# isinstance HTTPClientError: True
#
# Error: HTTP 401 from 'https://api.example.com/secure'
# retryable: False
# to_dict: {'error': 'auth_error', 'url': ..., 'status_code': 401, ...}
# isinstance HTTPClientError: True
Quick Reference
| Pattern | Syntax | Use When |
|---|---|---|
| Minimal exception | class MyError(Exception): pass | Message alone is enough |
| Rich exception | Custom __init__ with attributes + super().__init__(msg) | Callers need structured data |
| Readable string | Override __str__ | Human message differs from repr |
| Debug representation | Override __repr__ | Want full attribute dump in logs |
| Exception hierarchy | Base LibraryError → SpecificError | Building a library |
| Subclass built-in | class MyError(ValueError): pass | Error is a specialised form of a known category |
| Exception chaining | raise MyError(...) from original | Wrapping third-party exceptions |
| Suppress chain | raise MyError(...) from None | Hiding implementation details |
| Mixin exception | class MyError(TypeA, TypeB): pass | Error genuinely belongs to two categories |
| Retryable flag | self.retryable = True | Network/transient errors |
| JSON serialisation | def to_dict(self) -> dict | API responses, logging |
| Naming convention | Always end in Error (PEP 8) | All error exceptions |
Key Takeaways
- Always inherit from
Exception, neverBaseException- your domain errors must participate normally in the exception handling ecosystem - The minimal custom exception is one line:
class MyError(Exception): pass- add complexity only when callers need it - Custom
__init__with attributes makes exceptions machine-readable data carriers, not just human messages; always callsuper().__init__(human_message) - Use
raise NewError(...) from originalto chain exceptions - this preserves debugging context while presenting a clean domain API - Design exception hierarchies with a single library base exception so callers can choose between broad and specific catches
- Exception attributes should let code branch logically without parsing the message string - treat the message as display-only
- End all error exception names in
Error(PEP 8); a mixin pattern enables belonging to multiple categories simultaneously
