Custom Middleware in FastAPI: From Logging to Header Validation
In this blog, I will show you how to create custom middleware in FastAPI. Middleware is useful when you want to do something before or after a request is processed by your API.
We will learn how to:
- Log each request (method, path, time)
- Check if a custom header exists
- Return a response early if a condition is not met
Let’s get started!
✅ What Is Middleware?
Middleware is a function that runs before and after every request.
For example:
- Before: log the request or check a token
- After: modify the response or measure response time
In FastAPI, you can create middleware using the @app.middleware("http")
decorator.
Project Setup
Step 1: Install FastAPI and Uvicorn
pip install fastapi uvicorn
Step 2: Create a file main.py
Let’s create a basic FastAPI app and add a simple endpoint.
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def read_root():
return {"message": "Hello from FastAPI"}
Now we add middleware.
Middleware Example 1: Logging Requests
This middleware will:
- Log method and path
- Log how many milliseconds the request takes
import time
from fastapi import Request
@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) * 1000
print(f"{request.method} {request.url.path} completed in {process_time:.2f}ms")
return response
Now when you run the app and call the /
endpoint, the terminal will show logs like:
GET / completed in 0.42ms
Middleware Example 2: Check Custom Header
Sometimes you want to allow or deny requests based on a custom header, like an API key.
We will check if the request has the header x-api-key
and if the value is correct.
from fastapi.responses import JSONResponse
@app.middleware("http")
async def check_api_key(request: Request, call_next):
api_key = request.headers.get("x-api-key")
if api_key != "my_secret_key":
return JSONResponse(content={"error": "Unauthorized"}, status_code=401)
response = await call_next(request)
return response
Test with curl
:
✅ With correct key:
curl -H "x-api-key: my_secret_key" http://localhost:8000
❌ Without or wrong key:
curl http://localhost:8000
You will get:
{"error": "Unauthorized"}
Combine Both Middleware Functions
You can combine both logging and header checking in the same middleware or split them for cleaner code.
@app.middleware("http")
async def middleware_all(request: Request, call_next):
# Log
start_time = time.time()
# Check API key
api_key = request.headers.get("x-api-key")
if api_key != "my_secret_key":
return JSONResponse(content={"error": "Unauthorized"}, status_code=401)
response = await call_next(request)
# Log processing time
duration = (time.time() - start_time) * 1000
print(f"{request.method} {request.url.path} took {duration:.2f}ms")
return response
✅ Summary
- Middleware in FastAPI helps you run code before and after every request.
- You can use it for:
- Logging
- Authentication or header checking
- Measuring response time
- Modifying the response
- Be careful: middleware runs for all endpoints, so make sure your logic is efficient.
Thanks for reading!
I hope now you understand how to use custom middleware in FastAPI.