FastAPI Logging Middleware

In web application development, tracking what's happening with your application is crucial for debugging, performance monitoring, and security. This is where logging comes in - and FastAPI provides excellent tools to implement logging through middleware.

In this guide, we'll learn how to create and use logging middleware in your FastAPI applications to track requests, responses, and other important application events.

What is Logging Middleware?

Middleware in FastAPI is a component that processes requests before they reach your route handlers and responses before they're sent back to the client. Logging middleware specifically focuses on recording information about these requests and responses.

Think of logging middleware as a security camera for your application - it watches everything coming in and going out, recording details for later review.

Why Use Logging Middleware?

Debugging: Track down errors and issues in your application
Monitoring: Keep an eye on application performance and usage
Security: Detect suspicious activity or unauthorized access attempts
Analytics: Understand how users interact with your application

Basic Logging Setup in FastAPI

Let's start with a basic logging setup for a FastAPI application:

python
import logging
from fastapi import FastAPI

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

# Create a logger instance
logger = logging.getLogger(__name__)

# Create FastAPI app
app = FastAPI()

@app.get("/")
async def read_root():
    logger.info("Root endpoint was called")
    return {"Hello": "World"}

While this works, it requires you to add logging statements to every endpoint. A better approach is to create middleware that automatically logs all requests and responses.

Creating Custom Logging Middleware

Let's create a custom middleware that logs information about each request and response:

python
import time
import uuid
import logging
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware

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

class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        request_id = str(uuid.uuid4())
        start_time = time.time()
        
        # Log request info
        logger.info(f"Request {request_id} started: {request.method} {request.url.path}")
        
        # Process the request
        try:
            response = await call_next(request)
            
            # Calculate request processing time
            process_time = time.time() - start_time
            
            # Log response info
            logger.info(f"Request {request_id} completed: Status {response.status_code} (took {process_time:.3f}s)")
            
            return response
        except Exception as e:
            # Log exceptions
            logger.error(f"Request {request_id} failed: {str(e)}")
            raise e

# Create FastAPI app
app = FastAPI()

# Add the middleware to the app
app.add_middleware(LoggingMiddleware)

@app.get("/")
async def read_root():
    return {"Hello": "World"}

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

How It Works

We create a custom middleware class that inherits from BaseHTTPMiddleware
In the dispatch method, we:
- Generate a unique ID for each request
- Record the start time
- Log the incoming request
- Process the request
- Calculate processing time
- Log the response details
We add the middleware to our FastAPI application with app.add_middleware()

When you run this application and make requests, you'll see log entries like:

2023-07-12 14:32:45,123 - __main__ - INFO - Request 3a7c9f8d-6e7b-4a1d-8c9e-3f6d8a2b1c0e started: GET /
2023-07-12 14:32:45,178 - __main__ - INFO - Request 3a7c9f8d-6e7b-4a1d-8c9e-3f6d8a2b1c0e completed: Status 200 (took 0.055s)

Advanced Logging Middleware Features

Now let's enhance our middleware with more advanced features:

1. Logging Request and Response Bodies

python
class AdvancedLoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        request_id = str(uuid.uuid4())
        start_time = time.time()
        
        # Log request details
        logger.info(f"Request {request_id} started: {request.method} {request.url.path}")
        
        # Log request headers
        logger.debug(f"Request {request_id} headers: {request.headers}")
        
        # For logging request body (need to handle it specially since can only read body once)
        body = b""
        if request.method in ["POST", "PUT"]:
            body = await request.body()
            # Create a new request with the already-read body
            request = Request(
                scope=request.scope,
                receive=request._receive,
            )
            # Logging the request body - be careful with sensitive data!
            if len(body) > 0:
                logger.debug(f"Request {request_id} body: {body.decode()}")
        
        try:
            response = await call_next(request)
            
            process_time = time.time() - start_time
            
            # Log response details
            logger.info(f"Request {request_id} completed: Status {response.status_code} (took {process_time:.3f}s)")
            
            # We can't easily log the response body without modifying it
            # For that, we'd need a more complex approach with StreamingResponse
            
            return response
        except Exception as e:
            logger.error(f"Request {request_id} failed: {str(e)}")
            raise e

caution

Logging request and response bodies can expose sensitive information. Be very careful implementing this in production systems, especially for endpoints that handle authentication, personal data, or payment information.

2. Contextual Logging with Request ID

To trace requests throughout your application, you can set up a context variable to store the request ID:

python
import contextvars
from typing import Optional

# Create a context variable for request ID
request_id_ctx_var: contextvars.ContextVar[Optional[str]] = contextvars.ContextVar("request_id", default=None)

class ContextualLoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        request_id = str(uuid.uuid4())
        
        # Store the request ID in the context variable
        token = request_id_ctx_var.set(request_id)
        
        logger.info(f"Request {request_id} started: {request.method} {request.url.path}")
        
        try:
            response = await call_next(request)
            logger.info(f"Request {request_id} completed: Status {response.status_code}")
            return response
        except Exception as e:
            logger.error(f"Request {request_id} failed: {str(e)}")
            raise e
        finally:
            # Reset the context variable
            request_id_ctx_var.reset(token)

Now, anywhere in your application, you can access the current request ID:

python
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    request_id = request_id_ctx_var.get()
    logger.info(f"Processing item {item_id} for request {request_id}")
    return {"item_id": item_id}

Integrating with External Logging Systems

For production applications, you might want to send logs to external systems like ELK Stack (Elasticsearch, Logstash, Kibana), Datadog, or other monitoring solutions.

Here's an example using the python-json-logger package to format logs as JSON for easy integration:

python
import logging
from pythonjsonlogger import jsonlogger
from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware

# Configure JSON logging
logHandler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter("%(asctime)s %(name)s %(levelname)s %(message)s")
logHandler.setFormatter(formatter)
logger = logging.getLogger()
logger.addHandler(logHandler)
logger.setLevel(logging.INFO)

class JsonLoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        # Similar implementation as before, but logs will be in JSON format
        # ...

Real-World Example: Complete Application with Logging

Let's put everything together in a more complete example with structured logging to a file:

python
import time
import uuid
import logging
from logging.handlers import RotatingFileHandler
import json
from fastapi import FastAPI, Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware
from datetime import datetime

# Configure logging with rotation
log_formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
log_file = "app.log"
log_handler = RotatingFileHandler(log_file, maxBytes=10485760, backupCount=5)  # 10MB per file, keep 5 files
log_handler.setFormatter(log_formatter)

logger = logging.getLogger("app")
logger.setLevel(logging.INFO)
logger.addHandler(log_handler)
logger.addHandler(logging.StreamHandler())  # Also log to console

class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        request_id = str(uuid.uuid4())
        start_time = time.time()
        
        # Extract client information
        client_host = request.client.host if request.client else "unknown"
        user_agent = request.headers.get("user-agent", "unknown")
        
        # Create structured log entry
        log_data = {
            "request_id": request_id,
            "client_ip": client_host,
            "method": request.method,
            "path": request.url.path,
            "user_agent": user_agent,
            "timestamp": datetime.now().isoformat()
        }
        
        logger.info(f"Request started: {json.dumps(log_data)}")
        
        try:
            response = await call_next(request)
            
            process_time = time.time() - start_time
            
            # Update log with response data
            log_data.update({
                "status_code": response.status_code,
                "processing_time": f"{process_time:.3f}s"
            })
            
            logger.info(f"Request completed: {json.dumps(log_data)}")
            
            # Add custom header with request ID (useful for debugging)
            response.headers["X-Request-ID"] = request_id
            
            return response
        except Exception as e:
            log_data.update({
                "error": str(e),
                "processing_time": f"{time.time() - start_time:.3f}s"
            })
            logger.error(f"Request failed: {json.dumps(log_data)}")
            raise HTTPException(status_code=500, detail="Internal server error")

# Create FastAPI app
app = FastAPI(title="API with Advanced Logging")

# Add middleware
app.add_middleware(LoggingMiddleware)

@app.get("/")
async def read_root():
    return {"message": "Hello World"}

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id < 0:
        # This will be caught by our middleware and logged
        raise ValueError("Item ID must be positive")
    logger.info(f"Retrieved item {item_id}")
    return {"item_id": item_id, "name": f"Item {item_id}"}

@app.post("/items/")
async def create_item(item: dict):
    logger.info(f"Created new item: {json.dumps(item)}")
    return {"item": item, "created": True}

Running this application will log detailed information about each request and response, including:

Unique request IDs
Client IP addresses
User agent information
Request methods and paths
Response status codes
Processing times
Any errors that occur

Using Structured Logging for Analytics

The logs generated by our middleware can be analyzed to gain insights into your application's usage. For example:

Performance monitoring: Track slow endpoints by analyzing processing times
Error rate monitoring: Identify problematic endpoints by tracking error rates
Usage patterns: Understand how users interact with your API
Security analysis: Detect suspicious patterns of access

With tools like ELK Stack or Grafana/Loki, you can create dashboards to visualize this data.

Summary

In this guide, we've learned how to:

Create basic logging middleware in FastAPI
Implement advanced features like request ID tracking
Log request and response details
Format logs for external systems
Build a complete application with comprehensive logging

Proper logging is an essential part of any production application. By implementing logging middleware in your FastAPI applications, you gain visibility into your application's behavior, making it easier to debug issues, monitor performance, and ensure security.

Additional Resources

Exercises

Basic Exercise: Implement the basic logging middleware in a simple FastAPI application and observe the logs.
Intermediate Exercise: Extend the middleware to include more information, such as query parameters and request headers.
Advanced Exercise: Create a dashboard using tools like Grafana to visualize the log data from your application.
Expert Exercise: Implement rate limiting based on the data collected by your logging middleware to prevent abuse of your API.

Happy coding!

If you spot any mistakes on this website, please let me know at [email protected]. I’d greatly appreciate your feedback! :)

What is Logging Middleware?​

Why Use Logging Middleware?​

Basic Logging Setup in FastAPI​

Creating Custom Logging Middleware​

How It Works​

Advanced Logging Middleware Features​

1. Logging Request and Response Bodies​

2. Contextual Logging with Request ID​

Integrating with External Logging Systems​

Real-World Example: Complete Application with Logging​

Using Structured Logging for Analytics​

Summary​

Additional Resources​

Exercises​