FastAPI Complete Course Module 12: Exception Handling

Tutorial Rays FastAPI Complete Course FastAPI Complete Course Module 12: Exception Handling
  • June 19, 2026

FastAPI Complete Course Module 12: Exception Handling

AI Reading

Quick summary of this article

FastAPI provides a robust system for handling errors in API applications, from basic HTTP exceptions to custom error handling with global handlers. This module covers how to raise HTTP exceptions, create custom exception classes, implement global exception handlers, log errors effectively, and structure consistent API error responses for clients.

  • HTTPException is the built-in way to return standard error responses with status codes and messages, and you can add custom headers to error responses when needed.
  • Custom exception classes allow you to define specific error types with additional attributes like error codes and database names, making error handling more organized.
  • Global exception handlers catch all exceptions of a specific type and return standardized JSON responses, ensuring consistent error formatting across the entire application.
  • Logging errors with Python's logging module and appropriate severity levels is essential for debugging and monitoring, and you should always include tracebacks using exc_info=True.
  • Using Pydantic models to define structured error response schemas helps API consumers handle errors programmatically with consistent fields like error flag, message, error code, and details.

Introduction

In any professional FastAPI application, things will go wrong. A user might request a resource that doesn’t exist, submit invalid data, or your database might be temporarily unavailable. How you handle these errors determines the reliability and user-friendliness of your API. This is where FastAPI Exception Handling becomes essential.

FastAPI provides a robust system for managing errors, from simple HTTP exceptions to complex custom error handling with global handlers. In this chapter, you will learn how to raise HTTP exceptions, create custom exception classes, implement a global exception handler, log errors effectively, and structure API error responses for your clients.

By the end of this module, you will be able to build APIs that gracefully handle errors, provide meaningful feedback to users, and maintain clean logs for debugging. This is a critical skill for any job-oriented FastAPI developer.

HTTP Exceptions

The most common way to handle errors in FastAPI is by raising HTTPException. This is a built-in class that allows you to return standard HTTP error responses with a status code and a message.

Basic HTTPException Usage

Let’s start with a simple example. Imagine you have an endpoint that returns user details. If the user ID is not found, you should return a 404 error.

from fastapi import FastAPI, HTTPException

app = FastAPI()

# A dummy database of users
users_db = {
    1: {"name": "Alice", "email": "alice@example.com"},
    2: {"name": "Bob", "email": "bob@example.com"},
}

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")
    return users_db[user_id]

Explanation:

  • Line 1: We import HTTPException from FastAPI.
  • Lines 4-7: A simple dictionary acts as our database.
  • Line 10: The endpoint takes a user_id as a path parameter.
  • Lines 11-12: If the user ID does not exist in our database, we raise an HTTPException with status code 404 and a detail message.
  • Line 13: If the user exists, we return the user data.

Adding Headers to HTTPException

Sometimes you need to add custom headers to your error response, for example, to indicate a retry time or to provide a unique error ID.

from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 0:
        headers = {"X-Error-Code": "invalid_id", "Retry-After": "30"}
        raise HTTPException(
            status_code=400,
            detail="Item ID cannot be zero",
            headers=headers
        )
    return {"item_id": item_id}

Explanation:

  • Line 7: We define a custom header dictionary.
  • Lines 8-12: When raising the exception, we pass the headers parameter. FastAPI will include these headers in the response.

Common Status Codes for HTTPException

  • 400 Bad Request: Invalid input from the client.
  • 401 Unauthorized: Missing or invalid authentication.
  • 403 Forbidden: Authenticated but not allowed to access.
  • 404 Not Found: Resource does not exist.
  • 422 Unprocessable Entity: Validation error (FastAPI uses this for Pydantic validation).
  • 500 Internal Server Error: Unexpected server-side error.

Custom Exceptions

While HTTPException is great for simple cases, real-world applications often need more specific error types. Custom exceptions allow you to define your own error classes with additional attributes, making your code more organized and your error responses more informative.

Creating a Custom Exception Class

Let’s create a custom exception for handling database-related errors.

class DatabaseError(Exception):
    def __init__(self, message: str, error_code: str, db_name: str):
        self.message = message
        self.error_code = error_code
        self.db_name = db_name
        super().__init__(self.message)

Explanation:

  • Line 1: We define a custom exception class that inherits from Python’s built-in Exception.
  • Lines 2-5: The constructor accepts a message, an error_code (for internal tracking), and the db_name where the error occurred.
  • Line 6: We call the parent constructor with the message so it behaves like a standard exception.

Raising Custom Exceptions

Now, let’s use this custom exception in a simulated database operation.

from fastapi import FastAPI

app = FastAPI()

class DatabaseError(Exception):
    def __init__(self, message: str, error_code: str, db_name: str):
        self.message = message
        self.error_code = error_code
        self.db_name = db_name
        super().__init__(self.message)

def get_user_from_database(user_id: int):
    # Simulate a database connection failure
    if user_id == 999:
        raise DatabaseError(
            message="Connection to database failed",
            error_code="DB_CONN_ERR",
            db_name="users_db"
        )
    # Simulate a successful query
    return {"id": user_id, "name": "Alice"}

@app.get("/users/{user_id}")
async def read_user(user_id: int):
    user = get_user_from_database(user_id)
    return user

Explanation:

  • Lines 12-16: We simulate a database error when user_id is 999. We raise our custom DatabaseError with specific details.
  • Line 21: If the endpoint calls get_user_from_database with user_id 999, the custom exception is raised.

However, if you run this code and hit the endpoint with user_id 999, FastAPI will return a 500 Internal Server Error with a generic message. This is because FastAPI does not know how to handle our custom exception yet. We need an exception handler.

Global Exception Handler

A global exception handler allows you to catch all exceptions of a specific type (or all exceptions) and return a standardized response. This is crucial for maintaining consistent API error responses across your entire application.

Handling Custom Exceptions Globally

Let’s add a handler for our DatabaseError.

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

app = FastAPI()

class DatabaseError(Exception):
    def __init__(self, message: str, error_code: str, db_name: str):
        self.message = message
        self.error_code = error_code
        self.db_name = db_name
        super().__init__(self.message)

@app.exception_handler(DatabaseError)
async def database_error_handler(request: Request, exc: DatabaseError):
    return JSONResponse(
        status_code=500,
        content={
            "error": True,
            "message": exc.message,
            "error_code": exc.error_code,
            "db_name": exc.db_name,
            "type": "DatabaseError"
        }
    )

def get_user_from_database(user_id: int):
    if user_id == 999:
        raise DatabaseError(
            message="Connection to database failed",
            error_code="DB_CONN_ERR",
            db_name="users_db"
        )
    return {"id": user_id, "name": "Alice"}

@app.get("/users/{user_id}")
async def read_user(user_id: int):
    user = get_user_from_database(user_id)
    return user

Explanation:

  • Line 12: We use the @app.exception_handler() decorator to register a handler for DatabaseError.
  • Lines 13-21: The handler function takes a Request object and the exception instance. It returns a JSONResponse with a structured error payload.
  • Lines 15-20: The response includes an error flag, a human-readable message, an internal error_code, the db_name, and the type of error.

Now, when a DatabaseError is raised, FastAPI will catch it and return a clean JSON response instead of a generic 500 error.

Handling All Unhandled Exceptions

You can also create a catch-all handler for any exception that is not specifically handled.

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

app = FastAPI()

@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=500,
        content={
            "error": True,
            "message": "An unexpected error occurred. Please try again later.",
            "type": "InternalServerError"
        }
    )

Note: Be careful with a catch-all handler. It will override all other exception handlers if placed after them. Always place specific handlers before the general one.

Overriding FastAPI’s Default Exception Handlers

You can also override the default handlers for HTTPException and RequestValidationError to customize the response format.

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

app = FastAPI()

@app.exception_handler(HTTPException)
async def custom_http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": True,
            "message": exc.detail,
            "status_code": exc.status_code
        }
    )

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={
            "error": True,
            "message": "Validation failed",
            "details": exc.errors()
        }
    )

Logging Errors

Logging is essential for debugging and monitoring your application in production. FastAPI works seamlessly with Python’s built-in logging module. You should log errors with appropriate severity levels so you can track issues without overwhelming your log storage.

Setting Up Basic Logging

import logging
from fastapi import FastAPI, HTTPException

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    handlers=[
        logging.FileHandler("app.log"),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

app = FastAPI()

@app.get("/divide/{a}/{b}")
async def divide(a: float, b: float):
    if b == 0:
        logger.error("Division by zero attempted: a=%s, b=%s", a, b)
        raise HTTPException(status_code=400, detail="Cannot divide by zero")
    result = a / b
    logger.info("Division successful: %s / %s = %s", a, b, result)
    return {"result": result}

Explanation:

  • Lines 4-10: We configure logging to write to both a file (app.log) and the console. The format includes timestamp, logger name, severity level, and message.
  • Line 12: We create a logger instance for this module.
  • Lines 17-18: When division by zero is attempted, we log an error with the specific values.
  • Line 20: On success, we log an info message.

Logging in Exception Handlers

You should always log exceptions in your global handlers to capture stack traces.

import logging
import traceback
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

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

class DatabaseError(Exception):
    pass

@app.exception_handler(DatabaseError)
async def database_error_handler(request: Request, exc: DatabaseError):
    logger.error("DatabaseError occurred: %s", exc, exc_info=True)
    return JSONResponse(
        status_code=500,
        content={"error": True, "message": "Database operation failed"}
    )

@app.exception_handler(Exception)
async def general_handler(request: Request, exc: Exception):
    logger.error("Unhandled exception: %s", exc, exc_info=True)
    return JSONResponse(
        status_code=500,
        content={"error": True, "message": "Internal server error"}
    )

Key Points:

  • Use exc_info=True to include the full traceback in the log.
  • Use different log levels: debug for development, info for normal operations, warning for potential issues, error for failures, and critical for severe problems.
  • Never log sensitive information like passwords or API keys.

API Error Responses

Consistent error responses are crucial for API consumers. A well-structured error response makes it easy for frontend developers and other services to handle errors programmatically.

Structured Error Response Model

Use Pydantic models to define your error response schema.

from pydantic import BaseModel
from typing import Optional, List, Any

class ErrorResponse(BaseModel):
    error: bool = True
    message: str
    error_code: Optional[str] = None
    details: Optional[Any] = None
    type: str = "GenericError"

class ValidationErrorResponse(ErrorResponse):
    type: str = "ValidationError"
    details: List[dict] = []

Using the Error Response Model

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

app = FastAPI()

class ErrorResponse(BaseModel):
    error: bool = True
    message: str
    error_code: Optional[str] = None
    details: Optional[Any] = None
    type: str = "GenericError"

class ValidationErrorResponse(ErrorResponse):
    type: str = "ValidationError"
    details: List[dict] = []

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    response = ErrorResponse(
        message=exc.detail,
        error_code=f"HTTP_{exc.status_code}",
        type="HTTPException"
    )
    return JSONResponse(
        status_code=exc.status_code,
        content=response.model_dump()
    )

@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError):
    response = ValidationErrorResponse(
        message="Request validation failed",
        error_code="VALIDATION_ERR",
        details=exc.errors()
    )
    return JSONResponse(
        status_code=422,
        content=response.model_dump()
    )

Explanation:

  • Lines 8-14: We define a base ErrorResponse model with fields for error, message, error_code, details, and type.
  • Lines 16-18: A specialized ValidationErrorResponse extends the base model with a default type and a list of details.
  • Lines 20-27: In the HTTP exception handler, we create an ErrorResponse instance and convert it to a dictionary using model_dump().
  • Lines 29-36: For validation errors, we use the ValidationErrorResponse model to include detailed validation errors.

Practical Example: Complete Exception Handling System

Let’s put everything together into a single, practical example.

import logging
from typing import Optional, List, Any
from fastapi import FastAPI, Request, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("api.log"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

# Pydantic models for error responses
class ErrorResponse(BaseModel):
    error: bool = True
    message: str
    error_code: Optional[str] = None
    details: Optional[Any] = None
    type: str = "GenericError"

class ValidationErrorResponse(ErrorResponse):
    type: str = "ValidationError"
    details: List[dict] = []

# Custom exceptions
class DatabaseError(Exception):
    def __init__(self, message: str, error_code: str, db_name: str):
        self.message = message
        self.error_code = error_code
        self.db_name = db_name
        super().__init__(self.message)

class AuthenticationError(Exception):
    def __init__(self, message: str, user_id: Optional[int] = None):
        self.message = message
        self.user_id = user_id
        super().__init__(self.message)

# Initialize FastAPI app
app = FastAPI()

# Exception handlers
@app.exception_handler(DatabaseError)
async def database_error_handler(request: Request, exc: DatabaseError):
    logger.error("Database error in %s: %s", exc.db_name, exc.message, exc_info=True)
    response = ErrorResponse(
        message=exc.message,
        error_code=exc.error_code,
        type="DatabaseError"
    )
    return JSONResponse(status_code=500, content=response.model_dump())

@app.exception_handler(AuthenticationError)
async def auth_error_handler(request: Request, exc: AuthenticationError):
    logger.warning("Authentication failed for user %s: %s", exc.user_id, exc.message)
    response = ErrorResponse(
        message=exc.message,
        error_code="AUTH_ERR",
        type="AuthenticationError"
    )
    return JSONResponse(status_code=401, content=response.model_dump())

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    logger.info("HTTP exception: %s - %s", exc.status_code, exc.detail)
    response = ErrorResponse(
        message=exc.detail,
        error_code=f"HTTP_{exc.status_code}",
        type="HTTPException"
    )
    return JSONResponse(status_code=exc.status_code, content=response.model_dump())

@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError):
    logger.warning("Validation error: %s", exc.errors())
    response = ValidationErrorResponse(
        message="Request validation failed",
        error_code="VALIDATION_ERR",
        details=exc.errors()
    )
    return JSONResponse(status_code=422, content=response.model_dump())

@app.exception_handler(Exception)
async def general_handler(request: Request, exc: Exception):
    logger.critical("Unhandled exception: %s", exc, exc_info=True)
    response = ErrorResponse(
        message="An unexpected error occurred",
        error_code="INTERNAL_ERR",
        type="InternalServerError"
    )
    return JSONResponse(status_code=500, content=response.model_dump())

# Example endpoints
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    if user_id == 0:
        raise HTTPException(status_code=400, detail="Invalid user ID")
    if user_id == 999:
        raise DatabaseError(
            message="Could not connect to database",
            error_code="DB_CONN_ERR",
            db_name="users_db"
        )
    if user_id < 0:
        raise AuthenticationError(
            message="Invalid authentication token",
            user_id=user_id
        )
    return {"id": user_id, "name": "Alice"}

Common Mistakes

  1. Not logging exceptions: Many beginners raise exceptions but forget to log them. Always log errors with appropriate context.
  2. Exposing sensitive information: Avoid returning stack traces or internal details in production error responses. Log them, but return generic messages.
  3. Using generic HTTP status codes: Use specific status codes (404 for not found, 400 for bad request, etc.) instead of always returning 500.
  4. Inconsistent error response format: Always return the same structure (e.g., {error, message, error_code}) for all errors.
  5. Forgetting to handle validation errors: FastAPI automatically validates request data, but you should customize the validation error response for better user experience.
  6. Overusing catch-all handlers: A catch-all handler should be your last resort. Always try to handle specific exceptions first.

Practice Task

Create a small FastAPI application that manages a list of books. Implement the following:

  1. Create a custom exception called BookNotFoundError with fields for book_id and message.
  2. Create a custom exception called InvalidBookDataError with fields for field and message.
  3. Implement a global exception handler for both custom exceptions.
  4. Add a catch-all handler for any other exceptions.
  5. Log all errors with appropriate severity levels.
  6. Use a structured error response model (like ErrorResponse from the examples).
  7. Create endpoints:
    • GET /books/{book_id} – Returns a book or raises BookNotFoundError.
    • POST /books – Accepts a JSON body and validates that title is not empty. Raises InvalidBookDataError if validation fails.

Test your API using curl or the interactive docs at /docs. Verify that errors return consistent, structured JSON responses.

Summary

In this module, you learned the complete system of FastAPI Exception Handling. We started with the basic HTTPException and progressed to creating custom exception classes for specific error scenarios. You now know how to implement global exception handlers that catch and format errors consistently across your entire application.

We covered the importance of logging errors with context and severity levels, which is crucial for debugging in production. You also learned how to structure your API error responses using Pydantic models, ensuring that your API consumers always receive predictable, well-documented error payloads.

Key takeaways:

  • Use HTTPException for standard HTTP errors.
  • Create custom exception classes for domain-specific errors.
  • Register exception handlers with @app.exception_handler().
  • Always log errors with exc_info=True to capture stack traces.
  • Use Pydantic models to define consistent error response schemas.

FAQs

1. What is the difference between HTTPException and custom exceptions?

HTTPException is a built-in FastAPI class that automatically returns an HTTP response with a status code and detail message. Custom exceptions are Python classes you define for specific business logic errors. You must register handlers for custom exceptions to return proper HTTP responses.

2. Can I have multiple exception handlers for the same exception type?

No, only one handler can be registered per exception type. If you register multiple handlers for the same exception, the last one registered will be used.

3. How do I return custom headers in error responses?

You can pass a headers parameter to HTTPException when raising it. For custom exceptions, you can set headers directly in the JSONResponse object within your exception handler.

4. Should I always use a catch-all exception handler?

Yes, but only as a safety net. Always try to handle specific exceptions first. The catch-all handler should log the error and return a generic 500 response without exposing internal details.

5. How can I test my exception handlers?

You can use FastAPI’s TestClient to simulate requests that trigger exceptions. For example, call an endpoint with invalid data or a non-existent resource, and assert that the response status code and body match your expected error format.

You have now mastered exception handling in FastAPI. In the next module, Module 13: Middleware and CORS, you will learn how to process requests and responses globally using middleware, and how to configure Cross-Origin Resource Sharing (CORS) to allow your API to be accessed from different domains. These are essential skills for building production-ready APIs that integrate with frontend applications.

More Practical Examples

Let’s explore additional real-world scenarios where exception handling in FastAPI becomes essential. These examples go beyond basic HTTP exceptions and demonstrate how to handle edge cases gracefully.

Handling Database Connection Errors

When your API interacts with a database, connection failures can occur. Here’s how to handle them properly:

from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
import sqlite3

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    try:
        conn = sqlite3.connect("database.db")
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,))
        user = cursor.fetchone()
        conn.close()
        
        if user is None:
            raise HTTPException(status_code=404, detail="User not found")
        return {"user_id": user[0], "name": user[1], "email": user[2]}
    
    except sqlite3.OperationalError as e:
        # Handle database-specific errors
        error_detail = f"Database error: {str(e)}"
        return JSONResponse(
            status_code=500,
            content={"error": "Database connection failed", "detail": error_detail}
        )
    except Exception as e:
        # Catch-all for unexpected errors
        return JSONResponse(
            status_code=500,
            content={"error": "Internal server error", "detail": str(e)}
        )

This example shows how to catch database-specific exceptions separately from general exceptions. The sqlite3.OperationalError catches issues like missing tables or corrupted databases, while the general exception handler catches everything else. Notice we use JSONResponse directly to return structured error data.

Validation Errors with Pydantic

FastAPI automatically validates request data using Pydantic models. You can customize how validation errors are presented:

from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field

app = FastAPI()

class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=20)
    email: str = Field(..., pattern=r"^[w.-]+@[w.-]+.w+$")
    age: int = Field(..., ge=0, le=150)

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    errors = []
    for error in exc.errors():
        error_msg = {
            "field": " -> ".join(str(loc) for loc in error["loc"]),
            "message": error["msg"],
            "type": error["type"]
        }
        errors.append(error_msg)
    
    return JSONResponse(
        status_code=422,
        content={"error": "Validation failed", "details": errors}
    )

@app.post("/users/")
async def create_user(user: UserCreate):
    # If validation passes, this code runs
    return {"message": "User created successfully", "user": user}

This custom validation handler transforms the default Pydantic error format into a more user-friendly structure. Each error includes the field name, a human-readable message, and the error type. This is especially useful when building APIs for frontend applications that need clear error messages.

Handling External API Failures

When your FastAPI app calls external services, you need to handle timeouts and failures gracefully:

import httpx
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse

app = FastAPI()

@app.get("/weather/{city}")
async def get_weather(city: str):
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"https://api.weather.com/v1/{city}",
                timeout=10.0  # 10 seconds timeout
            )
            response.raise_for_status()  # Raises HTTPStatusError for 4xx/5xx
            return response.json()
    
    except httpx.TimeoutException:
        raise HTTPException(
            status_code=504,
            detail="Weather service timed out. Please try again later."
        )
    except httpx.HTTPStatusError as e:
        # Map external API errors to our own
        if e.response.status_code == 404:
            raise HTTPException(
                status_code=404,
                detail=f"City '{city}' not found"
            )
        raise HTTPException(
            status_code=502,
            detail="Weather service returned an error"
        )
    except httpx.RequestError as e:
        raise HTTPException(
            status_code=503,
            detail="Weather service is currently unavailable"
        )

This example demonstrates proper error handling when integrating with third-party APIs. Each exception type is handled separately to provide specific error messages. The timeout parameter prevents your API from hanging indefinitely if the external service is slow.

Class-Based Example

For larger applications, organizing exception handling into classes improves code maintainability. Here’s a class-based approach using FastAPI’s dependency injection:

from fastapi import FastAPI, Depends, HTTPException, Request
from fastapi.responses import JSONResponse
from typing import Optional

class DatabaseError(Exception):
    """Custom exception for database-related errors"""
    def __init__(self, message: str, status_code: int = 500):
        self.message = message
        self.status_code = status_code
        super().__init__(self.message)

class ValidationError(Exception):
    """Custom exception for data validation errors"""
    def __init__(self, message: str, field: Optional[str] = None):
        self.message = message
        self.field = field
        self.status_code = 400
        super().__init__(self.message)

class ExternalServiceError(Exception):
    """Custom exception for external service failures"""
    def __init__(self, service_name: str, message: str):
        self.service_name = service_name
        self.message = message
        self.status_code = 502
        super().__init__(self.message)

# Global exception handlers for custom exceptions
@app.exception_handler(DatabaseError)
async def database_error_handler(request: Request, exc: DatabaseError):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": "Database error",
            "detail": exc.message,
            "type": "DatabaseError"
        }
    )

@app.exception_handler(ValidationError)
async def validation_error_handler(request: Request, exc: ValidationError):
    content = {
        "error": "Validation error",
        "detail": exc.message,
        "type": "ValidationError"
    }
    if exc.field:
        content["field"] = exc.field
    return JSONResponse(status_code=exc.status_code, content=content)

@app.exception_handler(ExternalServiceError)
async def external_service_error_handler(request: Request, exc: ExternalServiceError):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": f"External service error: {exc.service_name}",
            "detail": exc.message,
            "type": "ExternalServiceError"
        }
    )

# Service class using custom exceptions
class UserService:
    def __init__(self, db_connection: str = "default"):
        self.db_connection = db_connection
    
    async def get_user(self, user_id: int):
        try:
            # Simulated database operation
            if user_id < 0:
                raise ValidationError(
                    message="User ID cannot be negative",
                    field="user_id"
                )
            
            # Simulate database lookup
            if user_id == 0:
                raise DatabaseError("User database is corrupted")
            
            # Simulate external service call
            if user_id == 999:
                raise ExternalServiceError(
                    service_name="EmailService",
                    message="Failed to fetch user preferences"
                )
            
            return {"id": user_id, "name": "John Doe", "email": "john@example.com"}
        
        except (DatabaseError, ValidationError, ExternalServiceError):
            # Re-raise custom exceptions to be handled by global handlers
            raise
        except Exception as e:
            # Convert unexpected errors to DatabaseError
            raise DatabaseError(f"Unexpected error: {str(e)}")

@app.get("/users/{user_id}")
async def get_user(user_id: int, service: UserService = Depends()):
    user = await service.get_user(user_id)
    return {"success": True, "data": user}

This class-based approach offers several advantages:

  • Separation of concerns: Custom exceptions are defined separately from business logic
  • Reusability: The UserService class can be used across multiple endpoints
  • Consistent error responses: Global handlers ensure all errors follow the same format
  • Type safety: Each exception carries specific information about the error

Step-by-Step Exercise

Let’s build a complete exception handling system for a simple task management API. Follow these steps to practice what you’ve learned:

Step 1: Project Setup

Create a new file called task_api.py and set up the basic FastAPI app:

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

app = FastAPI()

# In-memory task storage
tasks = {}
task_id_counter = 1

Step 2: Define Custom Exceptions

class TaskNotFoundError(Exception):
    def __init__(self, task_id: int):
        self.task_id = task_id
        self.message = f"Task with ID {task_id} not found"
        super().__init__(self.message)

class TaskValidationError(Exception):
    def __init__(self, message: str):
        self.message = message
        super().__init__(self.message)

Step 3: Create Global Exception Handlers

@app.exception_handler(TaskNotFoundError)
async def task_not_found_handler(request, exc: TaskNotFoundError):
    return JSONResponse(
        status_code=404,
        content={"error": "Task not found", "detail": exc.message}
    )

@app.exception_handler(TaskValidationError)
async def task_validation_handler(request, exc: TaskValidationError):
    return JSONResponse(
        status_code=400,
        content={"error": "Validation error", "detail": exc.message}
    )

@app.exception_handler(Exception)
async def general_exception_handler(request, exc: Exception):
    logging.error(f"Unhandled exception: {str(exc)}")
    return JSONResponse(
        status_code=500,
        content={"error": "Internal server error", "detail": "An unexpected error occurred"}
    )

Step 4: Implement API Endpoints

class TaskCreate(BaseModel):
    title: str
    description: Optional[str] = None
    completed: bool = False

class Task(BaseModel):
    id: int
    title: str
    description: Optional[str]
    completed: bool

@app.post("/tasks/", response_model=Task)
async def create_task(task: TaskCreate):
    global task_id_counter
    
    if not task.title.strip():
        raise TaskValidationError("Task title cannot be empty")
    
    new_task = Task(
        id=task_id_counter,
        title=task.title,
        description=task.description,
        completed=task.completed
    )
    tasks[task_id_counter] = new_task
    task_id_counter += 1
    return new_task

@app.get("/tasks/{task_id}", response_model=Task)
async def get_task(task_id: int):
    if task_id not in tasks:
        raise TaskNotFoundError(task_id)
    return tasks[task_id]

@app.get("/tasks/", response_model=List[Task])
async def list_tasks():
    return list(tasks.values())

@app.put("/tasks/{task_id}", response_model=Task)
async def update_task(task_id: int, task_update: TaskCreate):
    if task_id not in tasks:
        raise TaskNotFoundError(task_id)
    
    if not task_update.title.strip():
        raise TaskValidationError("Task title cannot be empty")
    
    updated_task = Task(
        id=task_id,
        title=task_update.title,
        description=task_update.description,
        completed=task_update.completed
    )
    tasks[task_id] = updated_task
    return updated_task

@app.delete("/tasks/{task_id}")
async def delete_task(task_id: int):
    if task_id not in tasks:
        raise TaskNotFoundError(task_id)
    
    del tasks[task_id]
    return {"message": f"Task {task_id} deleted successfully"}

Step 5: Test Your API

Run the application and test the endpoints:

# Start the server
uvicorn task_api:app --reload

# Test creating a task
curl -X POST "http://localhost:8000/tasks/" 
  -H "Content-Type: application/json" 
  -d '{"title": "Learn FastAPI", "description": "Study exception handling"}'

# Test getting a non-existent task (should return 404)
curl "http://localhost:8000/tasks/999"

# Test creating a task with empty title (should return 400)
curl -X POST "http://localhost:8000/tasks/" 
  -H "Content-Type: application/json" 
  -d '{"title": ""}'

Step 6: Add Logging

Enhance your error handling with proper logging:

import logging
from datetime import datetime

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('api_errors.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

@app.exception_handler(TaskNotFoundError)
async def task_not_found_handler(request, exc: TaskNotFoundError):
    logger.warning(f"Task not found: {exc.task_id} - IP: {request.client.host}")
    return JSONResponse(
        status_code=404,
        content={"error": "Task not found", "detail": exc.message}
    )

Interview and Job Use Cases

Exception handling in FastAPI is a common topic in technical interviews and real-world job scenarios. Here’s what you need to know:

Interview Questions You Might Face

  1. “How do you handle validation errors in FastAPI?”
    Explain that FastAPI automatically validates request data using Pydantic models, but you can customize the response using RequestValidationError exception handlers.
  2. “What’s the difference between HTTPException and custom exceptions?”
    HTTPException is built-in and returns standard HTTP error responses. Custom exceptions allow you to define application-specific errors with custom attributes and handling logic.
  3. “How would you handle database connection failures in a production API?”
    Discuss using try-except blocks around database operations, implementing retry logic, and returning appropriate HTTP status codes (503 for unavailable, 500 for unexpected errors).
  4. “What logging strategy would you implement for error tracking?”
    Mention using Python’s logging module with different levels (ERROR, WARNING, INFO), rotating file handlers, and integrating with monitoring tools like Sentry or ELK stack.

Real-World Job Scenarios

Scenario 1: E-commerce API
In an e-commerce application, you might need to handle:

  • Payment gateway failures (502 Bad Gateway)
  • Out-of-stock products (409 Conflict)
  • Invalid coupon codes (400 Bad Request)
  • User authentication errors (401 Unauthorized)

Scenario 2: Social Media Platform
Common exceptions include:

  • Rate limiting (429 Too Many Requests)
  • Content moderation failures (422 Unprocessable Entity)
  • File upload size exceeded (413 Payload Too Large)
  • Friend request already sent (409 Conflict)

Production Best Practices

# Example logging configuration for production
export LOG_LEVEL=ERROR
export LOG_FILE=/var/log/fastapi/app.log
export SENTRY_DSN=https://your-sentry-dsn@sentry.io/12345
# Production-ready exception handling
import sentry_sdk
from sentry_sdk.integrations.fastapi import FastApiIntegration

sentry_sdk.init(
    dsn="your-sentry-dsn",
    integrations=[FastApiIntegration()],
    traces_sample_rate=1.0
)

@app.exception_handler(Exception)
async def production_error_handler(request, exc: Exception):
    # Send to Sentry for monitoring
    sentry_sdk.capture_exception(exc)
    
    # Log the error
    logger.error(f"Unhandled exception: {str(exc)}", exc_info=True)
    
    # Don't expose internal details in production
    return JSONResponse(
        status_code=500,
        content={"error": "Internal server error", "request_id": request.state.request_id}
    )

Extra Beginner FAQs

1. What’s the difference between raise HTTPException and return JSONResponse?

HTTPException is a special exception that FastAPI catches and converts to a JSON response. It’s cleaner for most cases because it integrates with FastAPI’s exception handling system. JSONResponse is a direct response object that bypasses exception handling. Use HTTPException unless you need to return a response without triggering exception handlers.

2. Should I catch all exceptions in every endpoint?

No! That would make your code repetitive and hard to maintain. Instead, use global exception handlers for common errors and only catch specific exceptions in endpoints when you need special handling. The global handler acts as a safety net for unexpected errors.

3. How do I return different status codes for different errors?

Create custom exception classes that include a status_code attribute. Then, in your global handler, use that attribute to set the response status code. This keeps your endpoint code clean while allowing different error types to return different HTTP statuses.

4. What’s the best way to log errors without exposing sensitive data?

Use structured logging with Python’s logging module. Log the error type, timestamp, and a sanitized message. Never log passwords, API keys, or personal user data. Consider using a logging service like Sentry or Datadog that automatically filters sensitive information.

5. How do I test my exception handling code?

Use FastAPI’s TestClient to simulate different error scenarios. Create test cases that trigger each exception type and verify the response status code and body. This ensures your error handling works correctly before deploying to production.

# Example test for exception handling
from fastapi.testclient import TestClient

client = TestClient(app)

def test_task_not_found():
    response = client.get("/tasks/999")
    assert response.status_code == 404
    assert response.json()["error"] == "Task not found"

def test_empty_title():
    response = client.post("/tasks/", json={"title": ""})
    assert response.status_code == 400
    assert response.json()["error"] == "Validation error"

By mastering exception handling in FastAPI, you’ll build more reliable and maintainable APIs that gracefully handle errors and provide clear feedback to users and developers alike.

Leave a Reply

Your email address will not be published. Required fields are marked *