Middleware, Dependencies, and Error Handling

Chapter Outline

Chapter 3: Middleware, Dependencies, and Error Handling

In this chapter, you will learn:

  • What middleware is and how to use it in FastAPI.
  • How to apply dependencies for cleaner, reusable logic.
  • How to handle errors and exceptions in a structured way.
  • How to test error responses with Pytest.

We will extend our Todo API from Chapter 2 by:

  1. Adding middleware for request logging.
  2. Creating a dependency to simulate authentication.
  3. Implementing custom error handling for invalid operations.

3.1 Middleware in FastAPI

In FastAPI, a middleware is a function that runs for every request and response in your application. They provide a central point to perform actions that affect your entire application.

It sits in the "middle" the request-response cycle, allowing you to execute code both before and after the response is generated, but before it's sent back to the client.

3.1.1 Key functions of middleware

Middleware is used to handle cross-cutting concerns that apply globally, preventing you from having to repeat the same code every single endpoint. This is a common an well known design pattern used in Express.js, a popular NodeJS application framework.

3.1.2 Common use cases

  • Logging: Recodging details about each request, such as the path, method, and processing time.
  • CORS (Cross-Origin Resource Sharing): Adding appropriate CORS headers to allow requests from different origins.
  • Caching/Compression: Compressing large responses with GZip or adding custom or cache header headers.
  • Security: Rate limit based on IP addresses.

3.1.3 How it works

In FastAPI, you define a middleware by using a function with the @app.middleware("http") decorator. This function receives the incoming request and a call_next function. The call_next function passes the request down the chain to the intended path operation, and your middleware can then capture and modify the response before it is returned.

3.1.4 Example: Request logging middleware

The following code demonstrates a middleware that logs a request, method, path, and duration prior to sending the response back to the client.

fastapi_todo/main.py
1import time
2from fastapi import FastAPI, Request
3

4app = FastAPI()
5

6@app.middleware("http")
7async def log_requests(request: Request, call_next):
8    """Middleware to log request method, path, and execution time."""
9    start_time = time.time()
10    response = await call_next(request)
11    duration = time.time() - start_time
12    print(f"{request.method} {request.url.path} completed in {duration:.4f}s")
13    return response
14

15@app.get("/")
16async def read_root():
17    return {"message": "Hello FastAPI!"}

Run your app in a terminal, and issue the following command in another terminal:

bash
curl http://127.0.0.1:8000/

You should see an output similar to the following on the terminal window running your app:

bash
2025-10-29 12:52:10,837 - INFO - GET / completed in 0.0012s

The middleware generated the text GET / completed in 0.0012s in the printed log line.

3.1.5 Example: Adding a custom header

The following code demonstrates a middleware that adds a custom header, X-Processed-By-Middleware, to every request.

fastapi_todo/main.py
1import time
2from fastapi import FastAPI, Request
3

4app = FastAPI()
5

6@app.middleware("http")
7async def add_process_time_header(request: Request, call_next):
8    # This part runs BEFORE the request is processed by the path operation.
9    # The `request` object can be modified here.
10    request.state.start_time = time.time()
11    
12    # Pass the modified request to the next function in the chain
13    # (e.g., the endpoint).
14    response = await call_next(request)
15    
16    # This part runs AFTER the path operation has returned a response.
17    process_time = time.time() - request.state.start_time
18    response.headers["X-Processed-By-Middleware"] = f"{process_time:.4f}s"
19    
20    # Return the modified response.
21    return response
22

23@app.get("/")
24async def read_root():
25    return {"message": "Hello FastAPI!"}

If you issued the following command in the terminal:

bash
curl -i http://127.0.0.1:8000/

You should see an output similar to the following on the terminal window:

bash
1HTTP/1.1 200 OK
2date: Wed, 29 Oct 2025 18:57:37 GMT
3server: uvicorn
4content-length: 33
5content-type: application/json
6x-processed-by-middleware: 0.0010s
7

8{"message":"Hello FastAPI!"}

3.2 Dependencies in FastAPI

Dependencies are functions that can be injected into routes to provide shared logic, like authentication, DB connections, etc. In FastAPI, dependencies and middleware both address cross-cutting concerns, but they operate at different levels of granularity and serve distinct purposes.

While middleware affects the entire application, dependencies are specific to particular route handlers or groups of handlers, and they inject values diectly into your functions.

3.2.1 Key characteristics

  • Scoped: Dependencies are tied to a specific path operation, router, or the entire application, giving you fine-grained control.
  • Input-focused: They are primarily used for preparing input data, such as parsing requests headers, querying a database for a user object, or validating a token.
  • Integrated: The return value of a dependency is passed as a parameter directly into your endpoint function, making it a clean and clean and intuitive way to access shared resources.
  • Can interrupt: If a dependency raises an HTTPException, it stops the request and returns an error response immediately, preventing the path operation from running.

3.2.2 Common use cases

  • Authentication: Getting the current user from a token in the request header.
  • Database sessions: Providing a database session object to a route handler.
  • Validation: Checking permissions for business logic that must pass before the main logic runs.

3.2.3 Example: Authentication

Here we simulate a simple API key authentication:

fastapi_todo/main.py
1from fastapi import Header, HTTPException, Depends
2

3def get_api_key(x_api_key: str = Header(...)) -> str:
4    """Simple dependency to check for API key header."""
5    if x_api_key != "secret-key":
6        raise HTTPException(status_code=403, detail="Forbidden: Invalid API Key")
7    return x_api_key
8

9@app.get("/secure-todos", dependencies=[Depends(get_api_key)])
10def secure_list_todos() -> list[Todo]:
11    """List todos, but only if API key is valid."""
12    return todos

Explanation:

  • Any request to /secure-todos must send a header (case insensitive):
bash
X-API-Key: secret-key
  • Otherwise, it will return 403 Forbidden.
  • Dependencies reduce code duplication and enforce security rules.

3.3 Error Handling

FastAPI offers a powerful and flexible system for handling errors, from simple HTTP errors to complex custom exceptions. By providing a clear and consistent way to communicate failures, it ensures your API is robust and developer-friendly. FastAPI's error handling can be broken down into these primary mechanisms:

3.3.1 HTTPException

This is the most common way to raise HTTP-specific errors in FastAPI. It is a standard Python exception with extra information for APIs, allow you to specify the HTTP status code and a detail message. When you raise an HTTPException, FastAPI automatically catches it and returns an appropriate JSON response.

In an earlier example, we raised an HTTPException for a missing request header x_api_key with the value secret-key.

In the following example, if a requested item is not found, you can raise an HTTPException with a 404 Not Found status code.

fastapi_todo/main.py
1from fastapi import HTTPException
2

3@app.get("/todos/{todo_id}", response_model=TodoItem)
4def get_todo(todo_id: int) -> TodoItem:
5    """Retrieve a Todo item by ID."""
6    for todo in todos:
7        if todo.id == todo_id:
8            return todo
9    raise HTTPException(status_code=404, detail="Todo item not found")

3.3.2 Pydantic based

FastAPI's tight integration with Pydantic means it provides automatic data validation for requests. If a client dens invalid data (e.g., a string instead of a number), Pydantic raises a ValidationError, which FastAPI then transforms into a standard 422 Unprocessable Entity HTTP response.

fastapi_todo/main.py
1from pydantic import BaseModel, Field
2

3class TodoItem(BaseModel):
4    """Represents a Todo item."""
5    id: int
6    title: str = Field(..., min_length=1, max_length=100)
7    completed: bool = False
8

9# If a POST request to this endpoint sends a `id` that is not a number,
10# or non-string `title` FastAPI will automatically return a 422 error response.
11@app.post("/todos", response_model=TodoItem, status_code=201)
12def create_todo(todo: TodoCreate) -> TodoItem:
13    """Create a new Todo item."""
14    global next_id
15    new_todo = TodoItem(id=next_id, title=todo.title, completed=False)
16    todos.append(new_todo)
17    next_id += 1
18    return new_todo

Try the following command in a terminal window:

bash
curl -X POST http://127.0.0.1:8000/todos \
     -H "Content-Type: application/json" \
     -d '{"title": ""}'

You should see something like the following in the terminal as a reponse from the server:

json
{"error":"Validation failed","details":[{"type":"string_too_short","loc":["body","title"],"msg":"String should have at least 1 character","input":"","ctx":{"min_length":1}}]}

3.3.3 Global Exception Handlers

Register a handler for a specific exception type. When raised anywhere in the app, FastAPI will invoke this handler and return its result.

fastapi_todo/main.py
1from fastapi.responses import JSONResponse
2from fastapi.exception_handlers import RequestValidationError
3from fastapi.exceptions import RequestValidationError
4

5@app.exception_handler(RequestValidationError)
6async def validation_exception_handler(request, exc):
7    """Custom handler for validation errors."""
8    return JSONResponse(
9        status_code=422,
10        content={"error": "Validation failed", "details": exc.errors()},
11    )

You can also handle custom Python exceptions:

fastapi_todo/main.py
1class DatabaseError(Exception):
2    pass
3

4@app.exception_handler(DatabaseError)
5async def db_error_handler(request: Request, exc: DatabaseError):
6    return JSONResponse(status_code=500, content={"detail": "Database failure"})

3.3.4 Custom Exception Classes

Create your own exception class and corresponding handler for rich error information.

fastapi_todo/main.py
1class APIError(Exception):
2    def __init__(self, code: int, message: str, details: dict = None):
3        self.code = code
4        self.message = message
5        self.details = details or {}
6

7@app.exception_handler(APIError)
8async def api_error_handler(request: Request, exc: APIError):
9    return JSONResponse(
10        status_code=exc.code,
11        content={"message": exc.message, "details": exc.details}
12    )
13

14@app.get("/divide")
15def divide(a: float, b: float):
16    if b == 0:
17        raise APIError(
18            code=400,
19            message="Division by zero is not allowed",
20            details={"a": a, "b": b}
21        )
22    return {"result": a / b}

This tells FastAPI Whenever an APIError is raised anywhere in a request handler, use this function to produce the response. The handler converts the exception into a structured JSONResponse.

To try to trigger the custom error, you can use the following command:

bash
curl "http://127.0.0.1:8000/divide?a=10&b=0"

You should see an error response similar to the following in your console:

json
{"message":"Division by zero is not allowed","details":{"a":10.0,"b":0.0}}

3.4 Tests for Middleware, Dependencies, and Errors

python
1from fastapi.testclient import TestClient
2from fastapi_todo.main import app
3

4client = TestClient(app)
5

6

7def test_secure_todos_without_key() -> None:
8    response = client.get("/secure-todos")
9    assert response.status_code == 403
10    assert response.json() == {"error": "Forbidden: Invalid API Key"}
11

12

13def test_secure_todos_with_key() -> None:
14    response = client.get("/secure-todos", headers={"X-API-Key": "secret-key"})
15    assert response.status_code == 200
16    assert isinstance(response.json(), list)
17

18

19def test_validation_error() -> None:
20    response = client.post("/todos", json={})  # missing required "title"
21    assert response.status_code == 422
22    body = response.json()
23    assert body["error"] == "Validation failed"
24    assert "details" in body

Explanation:

  • Verifies that the dependency (API key) works correctly.
  • Confirms that validation errors return custom JSON responses.

3.5 Middleware Flow

graph TD C[Client] -->|Request| M[Middleware] M -->|Pre-process| H[Route Handler] H -->|Response| M M -->|Post-process| C

3.6 Dependency Injection

graph TD R[Route Handler] -->|Depends|get_api_key get_api_key -->|API Key| R

3.7 Further Reading

Check your understanding

Test your knowledge of Middleware, Dependencies, and Error Handling

Python
Web Dev

Feedback