mcpField
E
error-handling

error-handling

github.com/jiatasticDevelopment1088 installsUpdated 41w ago
Repository

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

  1. 01Raise errors near their source and catch at boundaries
  2. 02Define semantic domain exceptions
  3. 03Format errors with global handlers
  4. 04Preserve traceback context when wrapping errors
  5. 05Handle retries, cleanup, and background task failures

Use cases

  1. 01Centralize exception responses in a FastAPI service
  2. 02Wrap third-party SDK errors as domain errors
  3. 03Keep original tracebacks when adding context
  4. 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-handling

After installing, type/error-handlingin Claude Code to confirm it loaded.

Bundled files

References and scripts installed along with SKILL.md.

references/
asyncio.md
fastapi.md
pydantic.md
python.md

Details

From the project README

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:

SituationExample
Need to retrytenacity.retry() for transient failures
Need to transformWrap third-party SDK errors as domain errors
Need to clean upUse finally or context managers
Need to add contextraise DomainError(...) from original

Python + FastAPI Integration

LayerResponsibility
Service/DomainRaise domain exceptions (UserNotFoundError)
RoutesLet exceptions propagate (no try/except)
Exception HandlersTransform to HTTP responses
MiddlewareAdd 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

IssueCauseFix
Stack trace in responseNo generic handlerAdd @app.exception_handler(Exception)
Lost original errorMissing fromUse raise NewError() from original
Validation errors leakDefault handlerOverride RequestValidationError handler
Silent failuresSwallowed exceptionsLet exceptions propagate, handle at boundary

References

Comments

Markdown supported