About error-handling
A Python API error-handling guide based on letting exceptions propagate until an appropriate boundary. It is aimed at API developers, particularly those using FastAPI, and covers domain exceptions, centralized handlers, and preserving error context.
Key features
- 01Raise errors near their source and catch at boundaries
- 02Define semantic domain exceptions
- 03Format errors with global handlers
- 04Preserve traceback context when wrapping errors
- 05Handle retries, cleanup, and background task failures
Use cases
- 01Centralize exception responses in a FastAPI service
- 02Wrap third-party SDK errors as domain errors
- 03Keep original tracebacks when adding context
- 04Handle failures in background tasks
Install
Copy the command and run it in a terminal.
$
npx skills add https:/ / github.com/ jiatastic/ open-python-skills --skill error-handlingAfter installing, type/error-handlingin Claude Code to confirm it loaded.
Bundled files
References and scripts installed along with SKILL.md.
references/asyncio.mdfastapi.mdpydantic.mdpython.mdDetails
From the project READMEError 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
- Raise low, catch high - Throw exceptions where errors occur, handle at API boundaries
- Domain exceptions - Create semantic exceptions, not generic ones
- Global handlers - Use
@app.exception_handler()for centralized error formatting - No bare except - Always catch specific exceptions
- Preserve context - Use
raise ... from errorto 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:
| Situation | Example |
|---|---|
| Need to retry | tenacity.retry() for transient failures |
| Need to transform | Wrap third-party SDK errors as domain errors |
| Need to clean up | Use finally or context managers |
| Need to add context | raise DomainError(...) from original |
Python + FastAPI Integration
| Layer | Responsibility |
|---|---|
| Service/Domain | Raise domain exceptions (UserNotFoundError) |
| Routes | Let exceptions propagate (no try/except) |
| Exception Handlers | Transform to HTTP responses |
| Middleware | Add request context (request_id, timing) |
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
| Issue | Cause | Fix |
|---|---|---|
| Stack trace in response | No generic handler | Add @app.exception_handler(Exception) |
| Lost original error | Missing from | Use raise NewError() from original |
| Validation errors leak | Default handler | Override RequestValidationError handler |
| Silent failures | Swallowed exceptions | Let exceptions propagate, handle at boundary |
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
Comments
Markdown supported