FastAPI Timeout Middleware

When building web applications, managing request timeouts is crucial for maintaining application performance and reliability. In this tutorial, we'll explore how to implement timeout middleware in FastAPI applications, allowing you to automatically terminate requests that take too long to process.

What is Timeout Middleware?

Timeout middleware is a component that monitors the execution time of requests in your application. If a request exceeds a specified time limit, the middleware interrupts the execution and returns an appropriate response to the client, preventing resources from being tied up indefinitely by slow or hanging requests.

Why Use Timeout Middleware?

Resource Protection: Prevents long-running requests from consuming server resources
Improved User Experience: Ensures clients get timely responses rather than waiting indefinitely
Protection Against DoS Attacks: Helps mitigate attacks that try to exhaust server resources
Service Resilience: Improves overall application stability by limiting request duration

Basic Timeout Middleware Implementation

Let's start by creating a simple timeout middleware for FastAPI:

python
import asyncio
import time
from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware

class TimeoutMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, timeout_seconds=10):
        super().__init__(app)
        self.timeout_seconds = timeout_seconds
        
    async def dispatch(self, request: Request, call_next):
        try:
            # Start a timeout task
            start_time = time.time()
            
            # Create a task for the request processing
            response_task = asyncio.create_task(call_next(request))
            
            # Wait for the response with a timeout
            response = await asyncio.wait_for(response_task, timeout=self.timeout_seconds)
            
            # Calculate processing time
            process_time = time.time() - start_time
            response.headers["X-Process-Time"] = str(process_time)
            
            return response
            
        except asyncio.TimeoutError:
            return JSONResponse(
                status_code=504,
                content={"detail": f"Request timed out after {self.timeout_seconds} seconds"}
            )

Using the Timeout Middleware

Now let's see how to integrate the timeout middleware into your FastAPI application:

python
from fastapi import FastAPI

app = FastAPI()

# Add the timeout middleware to your application
app.add_middleware(TimeoutMiddleware, timeout_seconds=5)

@app.get("/")
async def root():
    return {"message": "This is a fast endpoint"}

@app.get("/slow")
async def slow_endpoint():
    # Simulate a slow operation
    await asyncio.sleep(10)
    return {"message": "This is a slow response"}

Testing the Timeout Middleware

When you run the application and test the endpoints:

The / endpoint will respond quickly with:

json
{
  "message": "This is a fast endpoint"
}

The /slow endpoint will timeout after 5 seconds with:

json
{
  "detail": "Request timed out after 5 seconds"
}

Advanced Timeout Middleware Features

Let's enhance our middleware to make it more robust and useful:

Configurable Timeout per Route

You can add path-specific timeouts by checking the request path:

python
class TimeoutMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, default_timeout=10, path_timeouts=None):
        super().__init__(app)
        self.default_timeout = default_timeout
        self.path_timeouts = path_timeouts or {}
        
    async def dispatch(self, request: Request, call_next):
        # Determine timeout based on path
        path = request.url.path
        timeout = self.path_timeouts.get(path, self.default_timeout)
        
        try:
            # Start a timeout task
            response_task = asyncio.create_task(call_next(request))
            response = await asyncio.wait_for(response_task, timeout=timeout)
            return response
            
        except asyncio.TimeoutError:
            return JSONResponse(
                status_code=504,
                content={"detail": f"Request timed out after {timeout} seconds"}
            )

Usage:

python
app.add_middleware(
    TimeoutMiddleware, 
    default_timeout=10,
    path_timeouts={
        "/fast": 2,
        "/slow": 15,
        "/api/data": 30
    }
)

Adding Logging

For better monitoring, you can add logging to your timeout middleware:

python
import logging

class TimeoutMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, timeout_seconds=10):
        super().__init__(app)
        self.timeout_seconds = timeout_seconds
        self.logger = logging.getLogger("timeout_middleware")
        
    async def dispatch(self, request: Request, call_next):
        try:
            start_time = time.time()
            response_task = asyncio.create_task(call_next(request))
            response = await asyncio.wait_for(response_task, timeout=self.timeout_seconds)
            
            process_time = time.time() - start_time
            self.logger.info(f"Request to {request.url} processed in {process_time:.2f}s")
            
            return response
            
        except asyncio.TimeoutError:
            self.logger.warning(f"Request to {request.url} timed out after {self.timeout_seconds}s")
            return JSONResponse(
                status_code=504,
                content={"detail": f"Request timed out after {self.timeout_seconds} seconds"}
            )

Real-World Example: API Rate Limiting with Timeout

Let's combine timeout middleware with rate limiting for a more comprehensive solution:

python
import time
from collections import defaultdict
from fastapi import FastAPI, Request, Response, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware

class RateLimitTimeoutMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, timeout_seconds=10, requests_limit=100, window_seconds=60):
        super().__init__(app)
        self.timeout_seconds = timeout_seconds
        self.requests_limit = requests_limit
        self.window_seconds = window_seconds
        self.request_counts = defaultdict(list)
        
    async def dispatch(self, request: Request, call_next):
        # Get client IP
        client_ip = request.client.host
        current_time = time.time()
        
        # Clean up old request timestamps
        self.request_counts[client_ip] = [
            timestamp for timestamp in self.request_counts[client_ip] 
            if current_time - timestamp < self.window_seconds
        ]
        
        # Check if rate limit exceeded
        if len(self.request_counts[client_ip]) >= self.requests_limit:
            return JSONResponse(
                status_code=429,
                content={"detail": "Rate limit exceeded. Please try again later."}
            )
        
        # Add current request timestamp
        self.request_counts[client_ip].append(current_time)
        
        # Process with timeout
        try:
            response_task = asyncio.create_task(call_next(request))
            response = await asyncio.wait_for(response_task, timeout=self.timeout_seconds)
            return response
            
        except asyncio.TimeoutError:
            return JSONResponse(
                status_code=504,
                content={"detail": f"Request timed out after {self.timeout_seconds} seconds"}
            )

Performance Considerations

When implementing timeout middleware, keep these considerations in mind:

Impact on Async Requests: The timeout mechanism works with FastAPI's async model but may add a small overhead.
Memory Usage: For long timeouts, ensure your server has sufficient memory to handle concurrent requests.
Task Cancellation: When a timeout occurs, ensure any background tasks or resources are properly cleaned up.

Best Practices

Set Appropriate Timeouts: Different endpoints may need different timeout values based on their processing requirements.
Monitor Timeout Occurrences: Regularly occurring timeouts could indicate performance problems in your application.
Add Proper Error Handling: Make sure clients receive informative error messages when timeouts occur.
Clean Up Resources: When a timeout occurs, make sure any open database connections or file handles are closed properly.

Summary

Timeout middleware is a powerful tool for maintaining the stability and responsiveness of your FastAPI applications. By implementing timeout controls:

You protect your application from resource exhaustion
You provide better user experience by ensuring timely responses
You improve overall application resilience

In production systems, timeout middleware should be combined with other fault-tolerance measures like rate limiting, circuit breakers, and proper error handling to create robust web services.

Additional Resources

Exercises

Modify the timeout middleware to exclude certain paths from timeout checks.
Implement a middleware that has different timeouts for different HTTP methods (GET vs. POST).
Create a middleware that logs detailed information about timed-out requests to a file.
Combine the timeout middleware with a circuit breaker pattern to make your application more resilient.
Implement a user-configurable timeout that can be set through request headers.

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

What is Timeout Middleware?​

Why Use Timeout Middleware?​

Basic Timeout Middleware Implementation​

Using the Timeout Middleware​

Testing the Timeout Middleware​

Advanced Timeout Middleware Features​

Configurable Timeout per Route​

Adding Logging​

Real-World Example: API Rate Limiting with Timeout​

Performance Considerations​

Best Practices​

Summary​

Additional Resources​

Exercises​