FastAPI is a powerful web framework that allows you to build robust APIs quickly. But as your application grows, you’ll need to implement advanced error handling and logging to maintain reliability and debug issues effectively. Let’s dive into some advanced techniques for custom error handling and logging in FastAPI.

First, let’s talk about error handling. FastAPI provides built-in exception handlers, but for more complex applications, you’ll want to create custom exception classes and handlers. This allows you to handle specific error scenarios in a more granular way.

Here’s an example of a custom exception class:

class CustomException(Exception):
    def __init__(self, message: str, status_code: int):
        self.message = message
        self.status_code = status_code

Now, let’s create a custom exception handler:

from fastapi import Request
from fastapi.responses import JSONResponse

@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"message": exc.message},
    )

This handler will catch any CustomException raised in your application and return a JSON response with the appropriate status code and message.

But what if you want to handle all unhandled exceptions? You can create a catch-all exception handler:

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=500,
        content={"message": "An unexpected error occurred"},
    )

This handler will catch any unhandled exception and return a generic error message. It’s a good practice to avoid exposing sensitive information in production, so keep your error messages user-friendly and non-specific.

Now, let’s talk about logging. Proper logging is crucial for debugging and monitoring your application. Python’s built-in logging module is powerful, but we can enhance it with some FastAPI-specific features.

First, let’s set up a basic logger:

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

To make our logs more informative, we can create a custom logging middleware that adds request details to each log entry:

from fastapi import FastAPI, Request
import time

app = FastAPI()

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    logger.info(f"{request.method} {request.url} took {process_time:.2f}s")
    return response

This middleware will log the HTTP method, URL, and processing time for each request.

But what if we want to log more detailed information about errors? We can enhance our exception handlers to include logging:

@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
    logger.error(f"CustomException: {exc.message}", exc_info=True)
    return JSONResponse(
        status_code=exc.status_code,
        content={"message": exc.message},
    )

This handler will log the full stack trace for CustomExceptions, which can be invaluable for debugging.

Now, let’s talk about structuring our logs. As your application grows, you’ll want to organize your logs in a way that makes them easy to search and analyze. One approach is to use structured logging:

import json

def log_info(message: str, **kwargs):
    log_entry = {
        "message": message,
        "level": "INFO",
        **kwargs
    }
    logger.info(json.dumps(log_entry))

This function allows you to log structured data that can be easily parsed by log analysis tools.

But what about performance? Logging can be I/O intensive, which might slow down your application. One way to mitigate this is to use asynchronous logging:

import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=3)

async def async_log(message: str, **kwargs):
    loop = asyncio.get_event_loop()
    await loop.run_in_executor(executor, log_info, message, **kwargs)

This function offloads the logging to a separate thread, allowing your main application to continue processing requests.

Now, let’s put it all together in a more complex example. We’ll create a FastAPI application that implements these advanced error handling and logging techniques:

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
import logging
import time
import json
import asyncio
from concurrent.futures import ThreadPoolExecutor

app = FastAPI()

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Set up async logging
executor = ThreadPoolExecutor(max_workers=3)

async def async_log(message: str, **kwargs):
    loop = asyncio.get_event_loop()
    await loop.run_in_executor(executor, log_info, message, **kwargs)

def log_info(message: str, **kwargs):
    log_entry = {
        "message": message,
        "level": "INFO",
        **kwargs
    }
    logger.info(json.dumps(log_entry))

# Custom exception
class CustomException(Exception):
    def __init__(self, message: str, status_code: int):
        self.message = message
        self.status_code = status_code

# Exception handlers
@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
    await async_log("CustomException occurred", message=exc.message, status_code=exc.status_code)
    return JSONResponse(
        status_code=exc.status_code,
        content={"message": exc.message},
    )

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    await async_log("Unhandled exception occurred", exc_info=True)
    return JSONResponse(
        status_code=500,
        content={"message": "An unexpected error occurred"},
    )

# Middleware for logging requests
@app.middleware("http")
async def log_requests(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    await async_log(
        "Request processed",
        method=request.method,
        url=str(request.url),
        process_time=f"{process_time:.2f}s"
    )
    return response

# Sample routes
@app.get("/")
async def root():
    return {"message": "Hello World"}

@app.get("/error")
async def trigger_error():
    raise CustomException("This is a custom error", status_code=400)

@app.get("/unhandled")
async def trigger_unhandled_error():
    raise ValueError("This is an unhandled error")

This example demonstrates how to implement advanced custom error handling and logging in FastAPI. It includes custom exceptions, exception handlers, request logging middleware, and asynchronous structured logging.

Remember, the key to effective error handling and logging is to provide enough information to debug issues without exposing sensitive data. Always sanitize your logs and error messages before sending them to the client.

As your application grows, you might want to consider using a more robust logging solution like ELK stack (Elasticsearch, Logstash, and Kibana) or a cloud-based logging service. These tools can help you aggregate, search, and visualize your logs more effectively.

Also, don’t forget about log rotation. As your application runs, it can generate a lot of log data. Implement a log rotation strategy to prevent your disk from filling up with old logs.

Lastly, remember that logging and error handling are not just about catching errors - they’re about providing visibility into your application’s behavior. Use them to track important events, monitor performance, and gain insights that can help you improve your application over time.

In conclusion, implementing advanced custom error handling and logging in FastAPI can significantly improve the reliability and maintainability of your application. By following these practices, you’ll be well-equipped to handle errors gracefully and debug issues effectively as your application scales.