Custom middleware creation in FastAPI - Mini Project: Build & Apply

Custom middleware creation

📖 Scenario: You are building a web API using FastAPI. You want to add a custom middleware that logs the time taken to process each request. This helps you understand how long your API takes to respond.

🎯 Goal: Create a custom middleware in FastAPI that measures and logs the time taken for each request.

📋 What You'll Learn

Create a FastAPI app instance

Define a custom middleware class that measures request processing time

Add the middleware to the FastAPI app

Create a simple GET endpoint to test the middleware

💡 Why This Matters

🌍 Real World

Custom middleware helps monitor and modify requests and responses in web APIs, useful for logging, authentication, and performance tracking.

💼 Career

Understanding middleware creation is essential for backend developers working with FastAPI or similar frameworks to build scalable and maintainable APIs.

1

Create FastAPI app instance

Write code to import FastAPI from fastapi and create an app instance called app.

FastAPI

# Your code here

Hint

Use from fastapi import FastAPI and then app = FastAPI().

2

Define a custom middleware class

Write a class called TimingMiddleware that inherits from BaseHTTPMiddleware imported from starlette.middleware.base. Inside, define an async method dispatch that takes request and call_next as parameters. Use time.perf_counter() to record start and end times around call_next(request). Calculate the duration and print it with the message "Request took {duration} seconds".

FastAPI

from fastapi import FastAPI

app = FastAPI()

# Your code here

Hint

Import BaseHTTPMiddleware and time. Define dispatch method with timing logic.

3

Add the middleware to the FastAPI app

Add the TimingMiddleware to the app using app.add_middleware(TimingMiddleware).

FastAPI

from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware
import time

app = FastAPI()

class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        start_time = time.perf_counter()
        response = await call_next(request)
        end_time = time.perf_counter()
        duration = end_time - start_time
        print(f"Request took {duration} seconds")
        return response

# Your code here

Hint

Use app.add_middleware(TimingMiddleware) to add the middleware.

4

Create a simple GET endpoint

Create a GET endpoint at path "/" using @app.get("/"). Define an async function called read_root that returns a dictionary {"message": "Hello World"}.

FastAPI

from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware
import time

app = FastAPI()

class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        start_time = time.perf_counter()
        response = await call_next(request)
        end_time = time.perf_counter()
        duration = end_time - start_time
        print(f"Request took {duration} seconds")
        return response

app.add_middleware(TimingMiddleware)

# Your code here

Hint

Use @app.get("/") decorator and define async def read_root() returning the dictionary.

Practice

1. What is the main purpose of creating custom middleware in FastAPI?

easy

A. To handle user authentication only

B. To define database models

C. To create HTML templates

D. To run code before and after each request is processed

Solution

1. Step 1: Understand middleware role

   Middleware runs code around request processing, before and after the main handler.

2. Step 2: Identify correct purpose

   Custom middleware is not for database or templates but for request/response handling.

3. Final Answer:

   To run code before and after each request is processed -> Option D

4. Quick Check:

   Middleware = pre/post request code [OK]

Hint: Middleware wraps requests to add extra processing [OK]

Common Mistakes:

• Confusing middleware with database or template code
• Thinking middleware only handles authentication
• Believing middleware runs only after requests

2. Which method must be overridden when creating a custom middleware class in FastAPI?

easy

A. dispatch

B. execute

C. process

D. handle_request

Solution

1. Step 1: Recall FastAPI middleware structure

   Custom middleware classes override the dispatch method to handle requests.

2. Step 2: Match method names

   Only dispatch is the correct method name; others are invalid in FastAPI middleware.

3. Final Answer:

   dispatch -> Option A

4. Quick Check:

   dispatch method overrides middleware behavior [OK]

Hint: dispatch handles request and response in middleware [OK]

Common Mistakes:

• Using incorrect method names like handle_request
• Confusing middleware methods with route handlers
• Forgetting to override dispatch method

3. Given this middleware code snippet, what will be printed when a request is processed?

class LogMiddleware:
    async def dispatch(self, request, call_next):
        print('Before request')
        response = await call_next(request)
        print('After request')
        return response

medium

A. Only Before request printed

B. Before request printed, then After request printed after response

C. Only After request printed

D. No output printed

Solution

1. Step 1: Analyze dispatch method flow

   Print 'Before request' runs before calling next handler, 'After request' runs after awaiting response.

2. Step 2: Understand async call_next behavior

   call_next processes the request and returns response; code after await runs after response is ready.

3. Final Answer:

   Before request printed, then After request printed after response -> Option B

4. Quick Check:

   Print before and after call_next = B [OK]

Hint: Code before await runs first, after await runs last [OK]

Common Mistakes:

• Thinking 'After request' prints before response
• Ignoring async await order
• Assuming only one print runs

4. Identify the error in this custom middleware code:

class MyMiddleware:
    async def dispatch(self, request):
        response = await call_next(request)
        return response

medium

A. Response must be created manually, not awaited

B. dispatch method should not be async

C. Missing call_next parameter in dispatch method

D. Middleware class must inherit from BaseHTTPMiddleware

Solution

1. Step 1: Check dispatch method signature

   dispatch must accept both request and call_next parameters to call next handler.

2. Step 2: Identify missing parameter

   Code lacks call_next parameter, causing runtime error when calling call_next(request).

3. Final Answer:

   Missing call_next parameter in dispatch method -> Option C

4. Quick Check:

   dispatch(request, call_next) required [OK]

Hint: dispatch needs call_next argument to forward requests [OK]

Common Mistakes:

• Omitting call_next parameter
• Making dispatch synchronous
• Not inheriting middleware base class (optional but recommended)

5. You want to create a middleware that adds a custom header 'X-Process-Time' showing how long the request took. Which code snippet correctly implements this?

hard

A. class TimerMiddleware:
async def dispatch(self, request, call_next):
start = time.time()
response = await call_next(request)
process_time = time.time() - start
response.headers['X-Process-Time'] = str(process_time)
return response

B. class TimerMiddleware:
async def dispatch(self, request):
start = time.time()
response = await call_next(request)
response.headers['X-Process-Time'] = str(time.time() - start)
return response

C. class TimerMiddleware:
def dispatch(self, request, call_next):
start = time.time()
response = call_next(request)
response.headers['X-Process-Time'] = str(time.time() - start)
return response

D. class TimerMiddleware:
async def dispatch(self, request, call_next):
response = await call_next(request)
response.headers['X-Process-Time'] = time.time()
return response

Solution

1. Step 1: Check dispatch method signature and async usage

   dispatch must be async and accept request and call_next parameters.

2. Step 2: Verify timing and header addition

   Start time before await call_next, calculate duration after, add header as string.

3. Step 3: Identify correct code snippet

   class TimerMiddleware: async def dispatch(self, request, call_next): start = time.time() response = await call_next(request) process_time = time.time() - start response.headers['X-Process-Time'] = str(process_time) return response correctly implements timing, async, and header addition.

4. Final Answer:

   The code snippet that correctly adds X-Process-Time header -> Option A

5. Quick Check:

   Async dispatch with timing and header = A [OK]

Hint: Measure time before and after await call_next, add header [OK]

Common Mistakes:

• Missing call_next parameter
• Not awaiting call_next
• Adding header before timing calculation
• Using synchronous dispatch method