Error Handling

Error Handling

Production-ready error handling for Python APIs using the Let it crash philosophy.

Design Philosophy

Let it crash - Don't be defensive. Let exceptions propagate naturally and handle them at boundaries.

# BAD - Too defensive, obscures errors
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    try:
        user = await user_service.get(user_id)
        if not user:
            raise HTTPException(404, "Not found")
        return user
    except DatabaseError as e:
        raise HTTPException(500, "Database error")
    except Exception as e:
        logger.exception("Unexpected error")
        raise HTTPException(500, "Internal error")

# GOOD - Let exceptions propagate, handle at boundary
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = await user_service.get(user_id)
    if not user:
        raise UserNotFoundError(user_id)
    return user

Core Principles

1. Raise low, catch high - Throw exceptions where errors occur, handle at API boundaries 2. Domain exceptions - Create semantic exceptions, not generic ones 3. Global handlers - Use @app.exception_handler() for centralized error formatting 4. No bare except - Always catch specific exceptions 5. Preserve context - Use raise ... from error to keep original traceback

Quick Start

1. Define Domain Exceptions

from enum import StrEnum

class ErrorCode(StrEnum):
    USER_NOT_FOUND = "user_not_found"
    INVALID_CREDENTIALS = "invalid_credentials"
    RATE_LIMITED = "rate_limited"

class DomainError(Exception):
    """Base exception for all domain errors."""
    def __init__(self, code: ErrorCode, message: str, status_code: int = 400):
        self.code = code
        self.message = message
        self.status_code = status_code
        super().__init__(message)

class UserNotFoundError(DomainError):
    def __init__(self, user_id: int):
        super().__init__(
            code=ErrorCode.USER_NOT_FOUND,
            message=f"User {user_id} not found",
            status_code=404
        )

2. Define Error Response Schema

from pydantic import BaseModel

class ErrorDetail(BaseModel):
    code: str
    message: str
    request_id: str | None = None

class ErrorResponse(BaseModel):
    error: ErrorDetail

3. Register Global Handlers

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()

@app.exception_handler(DomainError)
async def domain_error_handler(request: Request, exc: DomainError):
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": {"code": exc.code, "message": exc.message}}
    )

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": {"code": "http_error", "message": str(exc.detail)}}
    )

@app.exception_handler(RequestValidationError)
async def validation_error_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={"error": {"code": "validation_error", "message": "Invalid request"}}
    )

@app.exception_handler(Exception)
async def generic_error_handler(request: Request, exc: Exception):
    # Log full error internally
    logger.exception("Unhandled error")
    # Return safe message to client
    return JSONResponse(
        status_code=500,
        content={"error": {"code": "internal_error", "message": "Internal server error"}}
    )

4. Use in Routes

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = await user_service.get(user_id)
    if not user:
        raise UserNotFoundError(user_id)
    return user

When to Catch Exceptions

Only catch exceptions in these cases:

Python + FastAPI Integration

Common Patterns

Third-Party SDK Wrapping

import httpx
from tenacity import retry, stop_after_attempt, wait_exponential

class ExternalServiceError(DomainError):
    def __init__(self, service: str, original: Exception):
        super().__init__(
            code=ErrorCode.EXTERNAL_SERVICE_ERROR,
            message=f"{service} unavailable",
            status_code=503
        )
        self.__cause__ = original

@retry(stop=stop_after_attempt(3), wait=wait_exponential())
async def call_payment_api(data: dict):
    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            response = await client.post("https://api.payment.com/charge", json=data)
            response.raise_for_status()
            return response.json()
    except httpx.HTTPError as e:
        raise ExternalServiceError("Payment API", e) from e

Background Task Error Handling

from fastapi import BackgroundTasks

async def safe_background_task(task_func, *args, **kwargs):
    try:
        await task_func(*args, **kwargs)
    except Exception as e:
        logger.exception(f"Background task failed: {e}")
        # Optional: send to dead letter queue or alerting

@app.post("/orders")
async def create_order(order: Order, background_tasks: BackgroundTasks):
    result = await order_service.create(order)
    background_tasks.add_task(safe_background_task, send_confirmation_email, result.id)
    return result

Troubleshooting

References

• Python Patterns - Exception design, when to catch, SDK wrapping
• FastAPI Patterns - HTTPException, global handlers, middleware
• Pydantic Patterns - ValidationError, raise in validators
• Asyncio Patterns - TaskGroup, timeout, background tasks
• FastAPI Docs: Handling Errors
• Pydantic Docs: Error Handling

Asyncio Error Handling

Error handling patterns for async Python code, including TaskGroup, timeouts, and background tasks.

Basic try/except in async

Works the same as synchronous code:

async def fetch_user(user_id: int) -> User:
    try:
        return await db.users.get(user_id)
    except DatabaseError as e:
        raise UserServiceError("Database unavailable") from e

TaskGroup (Python 3.11+)

Basic Usage

TaskGroup provides structured concurrency with automatic cleanup:

import asyncio

async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(fetch_user(1))
        task2 = tg.create_task(fetch_user(2))
    
    # Both tasks completed successfully
    user1 = task1.result()
    user2 = task2.result()

Error Handling

If any task fails, all other tasks are cancelled:

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(fetch_user(1))      # Succeeds
            tg.create_task(failing_task())      # Raises error
            tg.create_task(slow_task())         # Gets cancelled
    except* ValueError as eg:
        # ExceptionGroup containing the ValueError
        for exc in eg.exceptions:
            print(f"Task failed: {exc}")

except* Syntax (Python 3.11+)

Handle multiple exceptions from TaskGroup:

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(task_raises_value_error())
            tg.create_task(task_raises_type_error())
    except* ValueError as eg:
        print(f"ValueError count: {len(eg.exceptions)}")
    except* TypeError as eg:
        print(f"TypeError count: {len(eg.exceptions)}")

Timeouts

asyncio.timeout (Python 3.11+)

import asyncio

async def fetch_with_timeout():
    try:
        async with asyncio.timeout(10.0):
            return await slow_operation()
    except TimeoutError:
        raise ExternalServiceError("Operation timed out")

asyncio.wait_for (Legacy)

async def fetch_with_timeout():
    try:
        return await asyncio.wait_for(slow_operation(), timeout=10.0)
    except asyncio.TimeoutError:
        raise ExternalServiceError("Operation timed out")

Nested Timeouts

Inner timeout should be shorter than outer:

async def complex_operation():
    async with asyncio.timeout(30.0):  # Overall timeout
        result1 = await fetch_data()
        
        async with asyncio.timeout(10.0):  # Per-operation timeout
            result2 = await process_data(result1)
        
        return result2

Background Tasks in FastAPI

The Problem

Background tasks run after the response is sent - exceptions are lost:

from fastapi import BackgroundTasks

@app.post("/orders")
async def create_order(order: Order, background_tasks: BackgroundTasks):
    result = await order_service.create(order)
    # If send_email fails, client never knows
    background_tasks.add_task(send_confirmation_email, result.id)
    return result

Solution: Wrapper Function

import logging

logger = logging.getLogger(__name__)

async def safe_background_task(
    task_func,
    *args,
    task_name: str = "background_task",
    **kwargs
):
    """Wrapper that catches and logs background task errors."""
    try:
        if asyncio.iscoroutinefunction(task_func):
            await task_func(*args, **kwargs)
        else:
            task_func(*args, **kwargs)
    except Exception as e:
        logger.exception(f"Background task '{task_name}' failed: {e}")
        # Optional: Send to error tracking, dead letter queue, etc.

@app.post("/orders")
async def create_order(order: Order, background_tasks: BackgroundTasks):
    result = await order_service.create(order)
    background_tasks.add_task(
        safe_background_task,
        send_confirmation_email,
        result.id,
        task_name="send_confirmation_email"
    )
    return result

Starlette Lifespan Events

Error in Startup

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    try:
        await database.connect()
        await cache.connect()
    except Exception as e:
        logger.exception("Startup failed")
        raise  # App won't start
    
    yield
    
    # Shutdown
    await database.disconnect()
    await cache.disconnect()

app = FastAPI(lifespan=lifespan)

Graceful Shutdown

@asynccontextmanager
async def lifespan(app: FastAPI):
    await startup()
    yield
    # Shutdown - don't raise, just log
    try:
        await database.disconnect()
    except Exception as e:
        logger.error(f"Error during database disconnect: {e}")
    
    try:
        await cache.disconnect()
    except Exception as e:
        logger.error(f"Error during cache disconnect: {e}")

CancelledError Handling

CancelledError is a BaseException (not Exception) - handle carefully:

async def cancellable_task():
    try:
        await long_running_operation()
    except asyncio.CancelledError:
        # Clean up resources
        await cleanup()
        raise  # Always re-raise CancelledError!

Don't Suppress CancelledError

# BAD - Breaks cancellation
async def bad_task():
    try:
        await operation()
    except asyncio.CancelledError:
        pass  # Swallows cancellation - task keeps running!

# GOOD - Re-raise after cleanup
async def good_task():
    try:
        await operation()
    except asyncio.CancelledError:
        await cleanup()
        raise

Common Pitfalls

Blocking the Event Loop

import asyncio

# BAD - Blocks event loop
async def bad_endpoint():
    time.sleep(10)  # Blocks!
    return {"status": "done"}

# GOOD - Use asyncio.sleep or run_in_executor
async def good_endpoint():
    await asyncio.sleep(10)  # Non-blocking
    return {"status": "done"}

# GOOD - For blocking I/O
async def good_endpoint():
    result = await asyncio.to_thread(blocking_io_operation)
    return result

Fire-and-Forget Tasks

# BAD - Task may be garbage collected
async def bad_handler():
    asyncio.create_task(background_work())  # May disappear!
    return {"status": "ok"}

# GOOD - Keep reference
background_tasks = set()

async def good_handler():
    task = asyncio.create_task(background_work())
    background_tasks.add(task)
    task.add_done_callback(background_tasks.discard)
    return {"status": "ok"}

Unchecked Task Exceptions

# BAD - Exception is lost
async def main():
    task = asyncio.create_task(failing_task())
    await asyncio.sleep(10)
    # Exception in failing_task is never seen

# GOOD - Check task result
async def main():
    task = asyncio.create_task(failing_task())
    await asyncio.sleep(10)
    if task.done():
        exc = task.exception()
        if exc:
            logger.error(f"Task failed: {exc}")

Concurrent Operations with Error Handling

asyncio.gather

async def fetch_all_users(user_ids: list[int]) -> list[User | None]:
    # return_exceptions=True - don't fail on first error
    results = await asyncio.gather(
        *[fetch_user(uid) for uid in user_ids],
        return_exceptions=True
    )
    
    users = []
    for uid, result in zip(user_ids, results):
        if isinstance(result, Exception):
            logger.error(f"Failed to fetch user {uid}: {result}")
            users.append(None)
        else:
            users.append(result)
    return users

TaskGroup vs gather

# Use TaskGroup when: all tasks must succeed
async with asyncio.TaskGroup() as tg:
    user = await tg.create_task(fetch_user(user_id))
    orders = await tg.create_task(fetch_orders(user_id))

# Use gather when: partial success is acceptable
results = await asyncio.gather(
    fetch_user(user_id),
    fetch_orders(user_id),
    return_exceptions=True
)

FastAPI Error Handling

FastAPI error handling using the global handler pattern - routes raise exceptions, handlers format responses.

Core Pattern

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()

# 1. Routes just raise - no try/except
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = await user_service.get(user_id)
    if not user:
        raise UserNotFoundError(user_id)
    return user

# 2. Global handlers format responses
@app.exception_handler(DomainError)
async def domain_error_handler(request: Request, exc: DomainError):
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": {"code": exc.code, "message": exc.message}}
    )

HTTPException

Basic Usage

from fastapi import HTTPException

@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return items[item_id]

With Custom Headers

raise HTTPException(
    status_code=401,
    detail="Invalid token",
    headers={"WWW-Authenticate": "Bearer"}
)

FastAPI vs Starlette HTTPException

from fastapi import HTTPException  # Accepts any JSON-able detail
from starlette.exceptions import HTTPException as StarletteHTTPException  # Only strings

# FastAPI's HTTPException allows dict detail
raise HTTPException(
    status_code=400,
    detail={"code": "invalid_input", "field": "email"}
)

# When registering handler, use Starlette's to catch both
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": {"code": "http_error", "message": str(exc.detail)}}
    )

Global Exception Handlers

Domain Exceptions

from enum import StrEnum

class ErrorCode(StrEnum):
    USER_NOT_FOUND = "user_not_found"
    PERMISSION_DENIED = "permission_denied"

class DomainError(Exception):
    def __init__(self, code: ErrorCode, message: str, status_code: int = 400):
        self.code = code
        self.message = message
        self.status_code = status_code

@app.exception_handler(DomainError)
async def domain_error_handler(request: Request, exc: DomainError):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": {
                "code": exc.code,
                "message": exc.message,
            }
        }
    )

RequestValidationError

Override default validation error format:

from fastapi.exceptions import RequestValidationError
from fastapi.encoders import jsonable_encoder

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    # Option 1: Simple message (hide details)
    return JSONResponse(
        status_code=422,
        content={"error": {"code": "validation_error", "message": "Invalid request"}}
    )

    # Option 2: Formatted errors
    errors = []
    for error in exc.errors():
        loc = ".".join(str(x) for x in error["loc"][1:])  # Skip "body"
        errors.append({"field": loc, "message": error["msg"]})
    return JSONResponse(
        status_code=422,
        content={"error": {"code": "validation_error", "errors": errors}}
    )

Generic Exception Handler

Catch-all for unexpected errors:

import logging

logger = logging.getLogger(__name__)

@app.exception_handler(Exception)
async def generic_exception_handler(request: Request, exc: Exception):
    # Log full error for debugging
    logger.exception("Unhandled exception", extra={
        "path": request.url.path,
        "method": request.method,
    })
    
    # Return safe message to client (hide internal details)
    return JSONResponse(
        status_code=500,
        content={"error": {"code": "internal_error", "message": "Internal server error"}}
    )

Middleware vs Exception Handlers

Adding Request Context via Middleware

import uuid
from starlette.middleware.base import BaseHTTPMiddleware

class RequestContextMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        request_id = str(uuid.uuid4())
        request.state.request_id = request_id
        
        response = await call_next(request)
        response.headers["X-Request-ID"] = request_id
        return response

app.add_middleware(RequestContextMiddleware)

# Use in exception handler
@app.exception_handler(DomainError)
async def domain_error_handler(request: Request, exc: DomainError):
    request_id = getattr(request.state, "request_id", None)
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": {
                "code": exc.code,
                "message": exc.message,
                "request_id": request_id,
            }
        }
    )

Dependency Injection Errors

Exceptions in dependencies propagate to handlers:

from fastapi import Depends

async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    user = await verify_token(token)
    if not user:
        raise AuthenticationError()  # Propagates to exception handler
    return user

@app.get("/me")
async def get_me(user: User = Depends(get_current_user)):
    return user  # Only reached if get_current_user succeeds

Reusing Default Handlers

Extend default behavior instead of replacing:

from fastapi.exception_handlers import (
    http_exception_handler,
    request_validation_exception_handler,
)

@app.exception_handler(StarletteHTTPException)
async def custom_http_exception_handler(request: Request, exc: StarletteHTTPException):
    # Add logging
    logger.warning(f"HTTP {exc.status_code}: {exc.detail}")
    # Then use default handler
    return await http_exception_handler(request, exc)

Security: Preventing Information Leakage

Don't Expose Stack Traces

# BAD - Exposes internals
@app.exception_handler(Exception)
async def bad_handler(request: Request, exc: Exception):
    import traceback
    return JSONResponse(
        status_code=500,
        content={"error": traceback.format_exc()}  # Never do this!
    )

# GOOD - Safe generic message
@app.exception_handler(Exception)
async def good_handler(request: Request, exc: Exception):
    logger.exception("Unhandled error")  # Log internally
    return JSONResponse(
        status_code=500,
        content={"error": {"code": "internal_error", "message": "Internal server error"}}
    )

Sanitize Validation Errors

@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError):
    # Don't include exc.body - may contain sensitive data
    return JSONResponse(
        status_code=422,
        content={"error": {"code": "validation_error", "message": "Invalid request"}}
    )

Error Response Schema

Consistent error format:

from pydantic import BaseModel
from typing import Any

class ErrorDetail(BaseModel):
    code: str
    message: str
    request_id: str | None = None
    errors: list[dict[str, Any]] | None = None  # For validation errors

class ErrorResponse(BaseModel):
    error: ErrorDetail

# Document in OpenAPI
@app.get("/users/{user_id}", responses={
    404: {"model": ErrorResponse, "description": "User not found"},
    500: {"model": ErrorResponse, "description": "Internal server error"},
})
async def get_user(user_id: int):
    ...

Pydantic Error Handling

Pydantic validation errors and how to raise errors in validators.

ValidationError Structure

When validation fails, Pydantic raises ValidationError with structured error data:

from pydantic import BaseModel, ValidationError

class User(BaseModel):
    name: str
    age: int

try:
    User(name="Alice", age="not-a-number")
except ValidationError as e:
    print(e.errors())

Each error contains:

Raising Errors in Validators

Use ValueError, Not ValidationError

Important: In validators, raise ValueError or AssertionError, not ValidationError directly.

from pydantic import BaseModel, field_validator

class User(BaseModel):
    password: str
    
    @field_validator("password")
    @classmethod
    def validate_password(cls, v: str) -> str:
        if len(v) < 8:
            raise ValueError("Password must be at least 8 characters")
        return v

PydanticCustomError for Rich Errors

Use PydanticCustomError to provide custom error types and context:

from pydantic import BaseModel, field_validator
from pydantic_core import PydanticCustomError

class User(BaseModel):
    password: str
    
    @field_validator("password")
    @classmethod
    def validate_password(cls, v: str) -> str:
        if len(v) < 8:
            raise PydanticCustomError(
                "password_too_short",  # Custom error type
                "Password must be at least {min_length} characters",
                {"min_length": 8}  # Context for message formatting
            )
        if not any(c.isupper() for c in v):
            raise PydanticCustomError(
                "password_no_uppercase",
                "Password must contain at least one uppercase letter",
                {}
            )
        return v

model_validator for Cross-Field Validation

from pydantic import BaseModel, model_validator

class UserCreate(BaseModel):
    password: str
    confirm_password: str
    
    @model_validator(mode="after")
    def passwords_match(self) -> "UserCreate":
        if self.password != self.confirm_password:
            raise ValueError("Passwords do not match")
        return self

Customizing Error Messages

Error Type Mapping

Map error types to custom messages:

from pydantic import ValidationError

CUSTOM_MESSAGES = {
    "int_parsing": "This field must be a number",
    "string_too_short": "This field is too short",
    "missing": "This field is required",
    "password_too_short": "Password is too weak",
}

def format_errors(exc: ValidationError) -> list[dict]:
    errors = []
    for error in exc.errors():
        error_type = error["type"]
        message = CUSTOM_MESSAGES.get(error_type, error["msg"])
        
        # Format with context if available
        if ctx := error.get("ctx"):
            message = message.format(**ctx)
        
        errors.append({
            "field": ".".join(str(x) for x in error["loc"]),
            "message": message,
        })
    return errors

In FastAPI Exception Handler

from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

app = FastAPI()

@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError):
    errors = []
    for error in exc.errors():
        loc = error["loc"]
        # Skip first element ("body", "query", "path")
        field = ".".join(str(x) for x in loc[1:]) if len(loc) > 1 else str(loc[0])
        errors.append({
            "field": field,
            "message": CUSTOM_MESSAGES.get(error["type"], error["msg"]),
        })
    
    return JSONResponse(
        status_code=422,
        content={"error": {"code": "validation_error", "errors": errors}}
    )

Error Location Format Conversion

Convert location tuple to dot notation:

def loc_to_dot(loc: tuple) -> str:
    """Convert ('items', 0, 'name') to 'items[0].name'"""
    parts = []
    for item in loc:
        if isinstance(item, int):
            parts.append(f"[{item}]")
        elif parts:
            parts.append(f".{item}")
        else:
            parts.append(item)
    return "".join(parts)

# Example usage
error_loc = ("items", 1, "price")
print(loc_to_dot(error_loc))  # "items[1].price"

Common Patterns

Conditional Validation

from pydantic import BaseModel, model_validator

class Order(BaseModel):
    order_type: str
    shipping_address: str | None = None
    
    @model_validator(mode="after")
    def validate_shipping(self) -> "Order":
        if self.order_type == "physical" and not self.shipping_address:
            raise ValueError("Shipping address required for physical orders")
        return self

Validation with External Data

from pydantic import BaseModel, field_validator

VALID_COUNTRIES = {"US", "CA", "UK", "DE", "FR"}

class Address(BaseModel):
    country: str
    
    @field_validator("country")
    @classmethod
    def validate_country(cls, v: str) -> str:
        if v.upper() not in VALID_COUNTRIES:
            raise ValueError(f"Country must be one of: {', '.join(VALID_COUNTRIES)}")
        return v.upper()

Anti-Patterns

Don't Raise ValidationError Directly

from pydantic import BaseModel, field_validator, ValidationError

class User(BaseModel):
    email: str
    
    # BAD - Don't raise ValidationError in validators
    @field_validator("email")
    @classmethod
    def validate_email(cls, v: str) -> str:
        if "@" not in v:
            raise ValidationError(...)  # Wrong!
        return v
    
    # GOOD - Raise ValueError
    @field_validator("email")
    @classmethod
    def validate_email(cls, v: str) -> str:
        if "@" not in v:
            raise ValueError("Invalid email format")
        return v

Don't Silently Fix Invalid Data

class User(BaseModel):
    age: int
    
    # BAD - Silently changes data
    @field_validator("age")
    @classmethod
    def validate_age(cls, v: int) -> int:
        if v < 0:
            return 0  # Silently fixes - user doesn't know
        return v
    
    # GOOD - Reject invalid data
    @field_validator("age")
    @classmethod
    def validate_age(cls, v: int) -> int:
        if v < 0:
            raise ValueError("Age cannot be negative")
        return v

Be Careful with Coercion

from pydantic import BaseModel

class Config(BaseModel):
    enabled: bool

# Pydantic coerces these to bool - may be unexpected
config = Config(enabled="false")  # enabled = True (non-empty string is truthy)
config = Config(enabled="0")      # enabled = True
config = Config(enabled=0)        # enabled = False

# Use strict mode if needed
from pydantic import ConfigDict

class StrictConfig(BaseModel):
    model_config = ConfigDict(strict=True)
    enabled: bool

StrictConfig(enabled="false")  # Raises ValidationError

Integration with FastAPI

from fastapi import FastAPI
from pydantic import BaseModel, field_validator

app = FastAPI()

class UserCreate(BaseModel):
    username: str
    password: str
    
    @field_validator("username")
    @classmethod
    def validate_username(cls, v: str) -> str:
        if len(v) < 3:
            raise ValueError("Username must be at least 3 characters")
        if not v.isalnum():
            raise ValueError("Username must be alphanumeric")
        return v.lower()

@app.post("/users")
async def create_user(user: UserCreate):
    # Validation happens automatically
    # Errors handled by global RequestValidationError handler
    return {"username": user.username}

Python Error Handling

Core Python error handling patterns following the Let it crash philosophy.

Design Philosophy

Raise Low, Catch High

Exceptions should be raised where errors occur and caught at application boundaries:

# In service layer - just raise
async def get_user(user_id: int) -> User:
    user = await db.users.get(user_id)
    if not user:
        raise UserNotFoundError(user_id)
    return user

# In route layer - let it propagate (no try/except needed)
@app.get("/users/{user_id}")
async def get_user_endpoint(user_id: int):
    return await user_service.get_user(user_id)

# At boundary - global handler catches and formats
@app.exception_handler(UserNotFoundError)
async def handle_user_not_found(request, exc):
    return JSONResponse(status_code=404, content={"error": exc.message})

When to Catch Exceptions

Only catch in these specific situations:

# GOOD - Catching to transform
try:
    response = await client.get(url)
except httpx.TimeoutException as e:
    raise ExternalServiceError("API timeout") from e

# GOOD - Catching to retry
@retry(stop=stop_after_attempt(3))
async def fetch_data():
    return await unreliable_api.get()

# GOOD - Catching to clean up (prefer context managers)
async with aiofiles.open("data.txt") as f:
    return await f.read()

# BAD - Catching just to log and re-raise
try:
    result = do_something()
except Exception as e:
    logger.error(f"Error: {e}")  # Redundant - global handler logs
    raise

Custom Exception Design

Exception Hierarchy

Create a hierarchy based on how callers will handle errors:

class AppError(Exception):
    """Base exception for application errors."""
    pass

class ValidationError(AppError):
    """Input validation failed."""
    pass

class NotFoundError(AppError):
    """Resource not found."""
    pass

class ExternalServiceError(AppError):
    """Third-party service failed."""
    def __init__(self, service: str, original: Exception | None = None):
        self.service = service
        super().__init__(f"{service} unavailable")
        if original:
            self.__cause__ = original

Storing Meaningful Attributes

Store data as attributes, not just strings:

# BAD - Only string message
class UserError(Exception):
    def __init__(self, message: str):
        super().__init__(message)

raise UserError("User 123 not found")  # Can't programmatically get user_id

# GOOD - Meaningful attributes
class UserNotFoundError(Exception):
    def __init__(self, user_id: int):
        self.user_id = user_id
        super().__init__(f"User {user_id} not found")

error = UserNotFoundError(123)
print(error.user_id)  # 123 - can use programmatically

Preserving Original Traceback

Always use from when re-raising:

# BAD - Loses original traceback
try:
    data = json.loads(raw)
except json.JSONDecodeError:
    raise ValidationError("Invalid JSON")

# GOOD - Preserves original traceback
try:
    data = json.loads(raw)
except json.JSONDecodeError as e:
    raise ValidationError("Invalid JSON") from e

Third-Party SDK Wrapping

Wrap at Integration Boundary

import stripe
from tenacity import retry, stop_after_attempt, wait_exponential

class PaymentError(AppError):
    """Payment processing failed."""
    pass

class PaymentDeclinedError(PaymentError):
    """Payment was declined."""
    def __init__(self, reason: str):
        self.reason = reason
        super().__init__(f"Payment declined: {reason}")

class PaymentServiceError(PaymentError):
    """Payment service unavailable."""
    pass

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
    retry=retry_if_exception_type(PaymentServiceError)
)
async def charge_card(amount: int, card_token: str) -> str:
    try:
        charge = stripe.Charge.create(amount=amount, source=card_token)
        return charge.id
    except stripe.error.CardError as e:
        raise PaymentDeclinedError(e.user_message) from e
    except stripe.error.RateLimitError as e:
        raise PaymentServiceError("Rate limited") from e
    except stripe.error.APIConnectionError as e:
        raise PaymentServiceError("Connection failed") from e
    except stripe.error.StripeError as e:
        raise PaymentError(f"Stripe error: {e}") from e

Timeout Handling

import asyncio
import httpx

async def fetch_with_timeout(url: str, timeout: float = 10.0) -> dict:
    """Fetch URL with explicit timeout."""
    try:
        async with httpx.AsyncClient(timeout=timeout) as client:
            response = await client.get(url)
            response.raise_for_status()
            return response.json()
    except httpx.TimeoutException as e:
        raise ExternalServiceError("Request timeout") from e
    except httpx.HTTPStatusError as e:
        raise ExternalServiceError(f"HTTP {e.response.status_code}") from e

# Or use asyncio.timeout (Python 3.11+)
async def fetch_with_asyncio_timeout(url: str) -> dict:
    try:
        async with asyncio.timeout(10.0):
            async with httpx.AsyncClient() as client:
                response = await client.get(url)
                return response.json()
    except TimeoutError:
        raise ExternalServiceError("Request timeout")

Anti-Patterns

Bare Except

# BAD - Catches everything including KeyboardInterrupt, SystemExit
try:
    do_something()
except:
    pass

# GOOD - Catch specific exceptions
try:
    do_something()
except ValueError as e:
    handle_value_error(e)

Catching Exception to Log and Re-raise

# BAD - Redundant, global handler should log
try:
    result = service.process()
except Exception as e:
    logger.exception("Error processing")
    raise

# GOOD - Just let it propagate
result = service.process()  # Global handler logs unhandled errors

Exception for Control Flow

# BAD - Using exception for normal flow
def find_user(user_id: int) -> User | None:
    try:
        return users[user_id]
    except KeyError:
        return None

# GOOD - Use .get() for expected missing keys
def find_user(user_id: int) -> User | None:
    return users.get(user_id)

Catching Too Broadly

# BAD - Catches unrelated errors (typos, logic bugs)
try:
    user = get_user(user_id)
    process_user(user)
except Exception:
    return None  # Hides bugs!

# GOOD - Catch only expected exceptions
try:
    user = get_user(user_id)
except UserNotFoundError:
    return None
process_user(user)  # Other errors propagate