FastAPI Error Handling and Custom Exception Handlers

from fastapi import FastAPI, HTTPException, status

app = FastAPI()

@app.get('/items/{item_id}')
async def get_item(item_id: int):
    if item_id > 100:
        # Raising HTTPException immediately stops the request flow
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f'Item {item_id} is out of stock or does not exist.',
            headers={'X-Forge-Error-Code': 'ERR_RESOURCE_NOT_FOUND'}
        )
    return {'item_id': item_id, 'status': 'available'}

from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse

app = FastAPI()

# Define a domain-specific exception
class InsufficientFundsError(Exception):
    def __init__(self, balance: float, amount: float):
        self.balance = balance
        self.amount = amount

# Register a global handler for this specific error type
@app.exception_handler(InsufficientFundsError)
async def insufficient_funds_handler(request: Request, exc: InsufficientFundsError):
    return JSONResponse(
        status_code=status.HTTP_402_PAYMENT_REQUIRED,
        content={
            'error_code': 'INSUFFICIENT_FUNDS',
            'message': 'Transaction declined due to low balance.',
            'meta': {
                'current_balance': exc.balance,
                'requested_amount': exc.amount,
                'deficit': round(exc.amount - exc.balance, 2)
            }
        }
    )

@app.post('/forge-pay/transfer')
async def process_transfer(amount: float):
    current_balance = 50.0
    if amount > current_balance:
        raise InsufficientFundsError(balance=current_balance, amount=amount)
    return {'status': 'success', 'transferred': amount}

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

app = FastAPI()

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    """
    Overwrites the default FastAPI 422 error to provide a flattened structure.
    Perfect for frontend forms that need simple 'field' -> 'message' mapping.
    """
    formatted_errors = []
    for error in exc.errors():
        formatted_errors.append({
            'location': error['loc'],
            'field': error['loc'][-1],
            'message': error['msg'],
            'type': error['type']
        })
        
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={'success': False, 'validation_errors': formatted_errors}
    )

import uuid
import logging
from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse

logger = logging.getLogger(__name__)
app = FastAPI()

@app.exception_handler(Exception)
async def generic_exception_handler(request: Request, exc: Exception):
    error_id = str(uuid.uuid4())[:8]
    # Log full traceback internally
    logger.exception(f"Unhandled error {error_id}: {exc}")
    # Return sanitized response
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={
            'error_code': 'INTERNAL_ERROR',
            'message': 'An unexpected error occurred. Please try again later.',
            'error_id': error_id
        }
    )

@app.get('/debug-crash')
async def crash():
    # Simulate an unhandled error
    _ = 1 / 0
    return {'ok': True}

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from structlog import get_logger

logger = get_logger()
app = FastAPI()

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    logger.warning(
        "HTTP error",
        status_code=exc.status_code,
        detail=exc.detail,
        path=str(request.url),
        method=request.method
    )
    return JSONResponse(
        status_code=exc.status_code,
        content={
            'error_code': f'HTTP_{exc.status_code}',
            'message': exc.detail
        },
        headers=exc.headers
    )

// io.thecodeforge — python tutorial

import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends, HTTPException

app = FastAPI()
logger = logging.getLogger(__name__)

# Simulate a DB connection that must be closed
class DatabaseConnection:
    async def close(self):
        logger.warning("Connection closed")

@asynccontextmanager
async def get_db():
    # Acquire resource before yield
    conn = DatabaseConnection()
    logger.info("Connection acquired")
    try:
        yield conn
    finally:
        # This runs even if the handler raises an exception
        await conn.close()
        logger.info("Cleanup executed")

@app.get("/users/{user_id}")
async def fetch_user(user_id: int, db: DatabaseConnection = Depends(get_db)):
    if user_id < 0:
        raise HTTPException(status_code=400, detail="user_id must be positive")
    # Simulate an unexpected crash
    raise RuntimeError("Unexpected database failure")

// io.thecodeforge — python tutorial

from pydantic import BaseModel
from typing import Optional, List
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse

app = FastAPI()

# Define the error contract — one model, every endpoint
class ApiError(BaseModel):
    code: str
    message: str
    details: Optional[List[dict]] = None

# Custom exceptions with built-in error codes
class PaymentError(Exception):
    def __init__(self, code: str, message: str, details: list = None):
        self.api_error = ApiError(code=code, message=message, details=details)

class RateLimitError(Exception):
    def __init__(self, retry_after: int):
        self.api_error = ApiError(
            code="RATE_LIMIT_EXCEEDED",
            message="Too many requests",
            details=[{"retry_after_seconds": retry_after}]
        )

# One handler to rule them all
@app.exception_handler(PaymentError)
@app.exception_handler(RateLimitError)
@app.exception_handler(HTTPException)
async def structured_error_handler(request: Request, exc):
    if hasattr(exc, 'api_error'):
        # Our custom exceptions carry the schema
        schema = exc.api_error
        return JSONResponse(status_code=422, content=schema.model_dump())
    # Fallback for plain HTTPException from FastAPI
    return JSONResponse(
        status_code=exc.status_code,
        content=ApiError(
            code="HTTP_ERROR",
            message=exc.detail
        ).model_dump()
    )

@app.get("/pay/{amount}")
async def pay(amount: int):
    if amount < 0:
        raise PaymentError(
            code="INVALID_AMOUNT",
            message="Payment amount cannot be negative",
            details=[{"field": "amount", "value": amount}]
        )
    return {"status": "ok"}