0
# Middleware
1
2
FastAPI provides comprehensive middleware support for processing HTTP requests and responses. Middleware components can modify requests before they reach route handlers and modify responses before they're sent to clients. FastAPI includes built-in middleware classes and supports custom middleware development.
3
4
## Capabilities
5
6
### Base Middleware Class
7
8
Base class for creating custom middleware components that process requests and responses.
9
10
```python { .api }
11
class Middleware:
12
def __init__(self, cls: type, **options: Any) -> None:
13
"""
14
Base middleware class for custom middleware.
15
16
Parameters:
17
- cls: Middleware class to instantiate
18
- options: Configuration options for the middleware
19
"""
20
self.cls = cls
21
self.options = options
22
```
23
24
### Middleware Decorator
25
26
Decorator method on FastAPI and APIRouter instances for adding middleware functions.
27
28
```python { .api }
29
def middleware(self, middleware_type: str) -> Callable[[Callable], Callable]:
30
"""
31
Decorator for adding middleware to the application.
32
33
Parameters:
34
- middleware_type: Type of middleware ("http" for HTTP middleware)
35
36
Returns:
37
Decorator function for middleware registration
38
"""
39
```
40
41
### CORS Middleware
42
43
Cross-Origin Resource Sharing middleware for handling cross-domain requests.
44
45
```python { .api }
46
class CORSMiddleware:
47
def __init__(
48
self,
49
app: ASGIApp,
50
allow_origins: List[str] = None,
51
allow_methods: List[str] = None,
52
allow_headers: List[str] = None,
53
allow_credentials: bool = False,
54
allow_origin_regex: str = None,
55
expose_headers: List[str] = None,
56
max_age: int = 600
57
) -> None:
58
"""
59
Cross-Origin Resource Sharing middleware.
60
61
Parameters:
62
- app: ASGI application to wrap
63
- allow_origins: List of allowed origin URLs
64
- allow_methods: List of allowed HTTP methods
65
- allow_headers: List of allowed request headers
66
- allow_credentials: Allow credentials in cross-origin requests
67
- allow_origin_regex: Regex pattern for allowed origins
68
- expose_headers: List of headers to expose to the browser
69
- max_age: Maximum age for preflight cache
70
"""
71
72
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
73
"""Process ASGI request with CORS handling."""
74
```
75
76
### GZip Middleware
77
78
Middleware for compressing HTTP responses using GZip compression.
79
80
```python { .api }
81
class GZipMiddleware:
82
def __init__(
83
self,
84
app: ASGIApp,
85
minimum_size: int = 500,
86
compresslevel: int = 9
87
) -> None:
88
"""
89
GZip compression middleware.
90
91
Parameters:
92
- app: ASGI application to wrap
93
- minimum_size: Minimum response size to compress (bytes)
94
- compresslevel: GZip compression level (1-9)
95
"""
96
97
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
98
"""Process ASGI request with GZip compression."""
99
```
100
101
### HTTPS Redirect Middleware
102
103
Middleware for enforcing HTTPS connections by redirecting HTTP requests.
104
105
```python { .api }
106
class HTTPSRedirectMiddleware:
107
def __init__(self, app: ASGIApp) -> None:
108
"""
109
HTTPS redirect enforcement middleware.
110
111
Parameters:
112
- app: ASGI application to wrap
113
"""
114
115
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
116
"""Process ASGI request with HTTPS redirect."""
117
```
118
119
### Trusted Host Middleware
120
121
Middleware for validating Host headers to prevent Host header attacks.
122
123
```python { .api }
124
class TrustedHostMiddleware:
125
def __init__(
126
self,
127
app: ASGIApp,
128
allowed_hosts: List[str] = None,
129
www_redirect: bool = True
130
) -> None:
131
"""
132
Trusted host validation middleware.
133
134
Parameters:
135
- app: ASGI application to wrap
136
- allowed_hosts: List of allowed host patterns
137
- www_redirect: Redirect www subdomain to non-www
138
"""
139
140
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
141
"""Process ASGI request with host validation."""
142
```
143
144
### WSGI Middleware
145
146
Middleware for mounting WSGI applications within ASGI applications.
147
148
```python { .api }
149
class WSGIMiddleware:
150
def __init__(self, app: ASGIApp, wsgi_app: WSGIApp) -> None:
151
"""
152
WSGI application mounting middleware.
153
154
Parameters:
155
- app: ASGI application to wrap
156
- wsgi_app: WSGI application to mount
157
"""
158
159
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
160
"""Process ASGI request with WSGI app mounting."""
161
```
162
163
### Custom Middleware Interface
164
165
Interface for creating custom HTTP middleware functions.
166
167
```python { .api }
168
async def custom_middleware_function(
169
request: Request,
170
call_next: Callable[[Request], Awaitable[Response]]
171
) -> Response:
172
"""
173
Custom middleware function signature.
174
175
Parameters:
176
- request: HTTP request object
177
- call_next: Function to call next middleware/route handler
178
179
Returns:
180
Response object (potentially modified)
181
"""
182
```
183
184
## Usage Examples
185
186
### Basic Custom Middleware
187
188
```python
189
import time
190
from fastapi import FastAPI, Request
191
192
app = FastAPI()
193
194
@app.middleware("http")
195
async def add_process_time_header(request: Request, call_next):
196
start_time = time.time()
197
response = await call_next(request)
198
process_time = time.time() - start_time
199
response.headers["X-Process-Time"] = str(process_time)
200
return response
201
202
@app.get("/")
203
async def read_main():
204
return {"message": "Hello World"}
205
```
206
207
### CORS Middleware Configuration
208
209
```python
210
from fastapi import FastAPI
211
from fastapi.middleware.cors import CORSMiddleware
212
213
app = FastAPI()
214
215
app.add_middleware(
216
CORSMiddleware,
217
allow_origins=["http://localhost:3000", "https://myapp.com"],
218
allow_credentials=True,
219
allow_methods=["GET", "POST", "PUT", "DELETE"],
220
allow_headers=["*"],
221
expose_headers=["X-Custom-Header"],
222
max_age=600
223
)
224
225
@app.get("/api/data")
226
async def get_data():
227
return {"data": "This endpoint supports CORS"}
228
```
229
230
### GZip Compression Middleware
231
232
```python
233
from fastapi import FastAPI
234
from fastapi.middleware.gzip import GZipMiddleware
235
236
app = FastAPI()
237
238
# Enable GZip compression for responses larger than 1000 bytes
239
app.add_middleware(GZipMiddleware, minimum_size=1000)
240
241
@app.get("/large-data")
242
async def get_large_data():
243
# Return large response that will be compressed
244
return {"data": "x" * 2000, "message": "This response will be compressed"}
245
```
246
247
### HTTPS Redirect Middleware
248
249
```python
250
from fastapi import FastAPI
251
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware
252
253
app = FastAPI()
254
255
# Redirect all HTTP requests to HTTPS
256
app.add_middleware(HTTPSRedirectMiddleware)
257
258
@app.get("/secure-endpoint")
259
async def secure_endpoint():
260
return {"message": "This endpoint requires HTTPS"}
261
```
262
263
### Trusted Host Middleware
264
265
```python
266
from fastapi import FastAPI
267
from fastapi.middleware.trustedhost import TrustedHostMiddleware
268
269
app = FastAPI()
270
271
# Only allow requests from specific hosts
272
app.add_middleware(
273
TrustedHostMiddleware,
274
allowed_hosts=["example.com", "*.example.com", "localhost"]
275
)
276
277
@app.get("/")
278
async def read_main():
279
return {"message": "Request from trusted host"}
280
```
281
282
### Authentication Middleware
283
284
```python
285
import jwt
286
from fastapi import FastAPI, Request, HTTPException
287
from fastapi.responses import JSONResponse
288
289
app = FastAPI()
290
291
SECRET_KEY = "your-secret-key"
292
ALGORITHM = "HS256"
293
294
@app.middleware("http")
295
async def authenticate_request(request: Request, call_next):
296
# Skip authentication for certain paths
297
if request.url.path in ["/login", "/docs", "/openapi.json"]:
298
response = await call_next(request)
299
return response
300
301
# Extract token from Authorization header
302
auth_header = request.headers.get("Authorization")
303
if not auth_header or not auth_header.startswith("Bearer "):
304
return JSONResponse(
305
status_code=401,
306
content={"error": "Missing or invalid authorization header"}
307
)
308
309
token = auth_header.replace("Bearer ", "")
310
311
try:
312
# Verify JWT token
313
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
314
request.state.user = payload
315
except jwt.InvalidTokenError:
316
return JSONResponse(
317
status_code=401,
318
content={"error": "Invalid token"}
319
)
320
321
response = await call_next(request)
322
return response
323
324
@app.get("/protected")
325
async def protected_route(request: Request):
326
return {"message": f"Hello {request.state.user['sub']}"}
327
```
328
329
### Logging Middleware
330
331
```python
332
import logging
333
import time
334
from fastapi import FastAPI, Request
335
336
# Configure logging
337
logging.basicConfig(level=logging.INFO)
338
logger = logging.getLogger(__name__)
339
340
app = FastAPI()
341
342
@app.middleware("http")
343
async def log_requests(request: Request, call_next):
344
start_time = time.time()
345
346
# Log request
347
logger.info(
348
f"Request: {request.method} {request.url.path}",
349
extra={
350
"method": request.method,
351
"path": request.url.path,
352
"query_params": str(request.query_params),
353
"client": request.client.host if request.client else None
354
}
355
)
356
357
# Process request
358
response = await call_next(request)
359
360
# Log response
361
process_time = time.time() - start_time
362
logger.info(
363
f"Response: {response.status_code} in {process_time:.4f}s",
364
extra={
365
"status_code": response.status_code,
366
"process_time": process_time,
367
"path": request.url.path
368
}
369
)
370
371
return response
372
373
@app.get("/")
374
async def read_main():
375
return {"message": "Hello World"}
376
```
377
378
### Rate Limiting Middleware
379
380
```python
381
import time
382
from collections import defaultdict
383
from fastapi import FastAPI, Request, HTTPException
384
385
app = FastAPI()
386
387
# Simple in-memory rate limiter
388
rate_limiter = defaultdict(list)
389
RATE_LIMIT = 10 # requests per minute
390
RATE_WINDOW = 60 # seconds
391
392
@app.middleware("http")
393
async def rate_limit_middleware(request: Request, call_next):
394
client_ip = request.client.host if request.client else "unknown"
395
current_time = time.time()
396
397
# Clean old requests
398
rate_limiter[client_ip] = [
399
req_time for req_time in rate_limiter[client_ip]
400
if current_time - req_time < RATE_WINDOW
401
]
402
403
# Check rate limit
404
if len(rate_limiter[client_ip]) >= RATE_LIMIT:
405
raise HTTPException(
406
status_code=429,
407
detail="Rate limit exceeded",
408
headers={"Retry-After": str(RATE_WINDOW)}
409
)
410
411
# Add current request
412
rate_limiter[client_ip].append(current_time)
413
414
response = await call_next(request)
415
return response
416
417
@app.get("/")
418
async def read_main():
419
return {"message": "Hello World"}
420
```
421
422
### Error Handling Middleware
423
424
```python
425
import traceback
426
from fastapi import FastAPI, Request, HTTPException
427
from fastapi.responses import JSONResponse
428
429
app = FastAPI()
430
431
@app.middleware("http")
432
async def catch_exceptions_middleware(request: Request, call_next):
433
try:
434
response = await call_next(request)
435
return response
436
except HTTPException:
437
# Re-raise HTTPExceptions to be handled by FastAPI
438
raise
439
except Exception as e:
440
# Handle unexpected exceptions
441
error_id = str(hash(str(e) + str(time.time())))
442
443
# Log the full traceback
444
logger.error(
445
f"Unhandled exception {error_id}: {str(e)}",
446
extra={
447
"error_id": error_id,
448
"path": request.url.path,
449
"method": request.method,
450
"traceback": traceback.format_exc()
451
}
452
)
453
454
return JSONResponse(
455
status_code=500,
456
content={
457
"error": "Internal server error",
458
"error_id": error_id,
459
"message": "An unexpected error occurred"
460
}
461
)
462
463
@app.get("/error")
464
async def trigger_error():
465
raise ValueError("This is a test error")
466
467
@app.get("/")
468
async def read_main():
469
return {"message": "Hello World"}
470
```
471
472
### Multiple Middleware Stack
473
474
```python
475
from fastapi import FastAPI, Request
476
from fastapi.middleware.cors import CORSMiddleware
477
from fastapi.middleware.gzip import GZipMiddleware
478
import time
479
480
app = FastAPI()
481
482
# Add multiple middleware in order
483
# Note: Middleware is executed in reverse order of addition
484
485
# 1. GZip (executed last - compresses final response)
486
app.add_middleware(GZipMiddleware, minimum_size=1000)
487
488
# 2. CORS (executed second to last)
489
app.add_middleware(
490
CORSMiddleware,
491
allow_origins=["*"],
492
allow_credentials=True,
493
allow_methods=["*"],
494
allow_headers=["*"],
495
)
496
497
# 3. Custom timing middleware (executed first)
498
@app.middleware("http")
499
async def add_timing_header(request: Request, call_next):
500
start_time = time.time()
501
response = await call_next(request)
502
process_time = time.time() - start_time
503
response.headers["X-Process-Time"] = str(process_time)
504
return response
505
506
@app.get("/")
507
async def read_main():
508
return {"message": "Hello World", "data": "x" * 1500} # Large response for GZip
509
```
510
511
### Conditional Middleware
512
513
```python
514
from fastapi import FastAPI, Request
515
import os
516
517
app = FastAPI()
518
519
# Only add CORS middleware in development
520
if os.getenv("ENVIRONMENT") == "development":
521
from fastapi.middleware.cors import CORSMiddleware
522
app.add_middleware(
523
CORSMiddleware,
524
allow_origins=["*"],
525
allow_credentials=True,
526
allow_methods=["*"],
527
allow_headers=["*"],
528
)
529
530
# Security middleware for production
531
if os.getenv("ENVIRONMENT") == "production":
532
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware
533
from fastapi.middleware.trustedhost import TrustedHostMiddleware
534
535
app.add_middleware(HTTPSRedirectMiddleware)
536
app.add_middleware(
537
TrustedHostMiddleware,
538
allowed_hosts=["myapp.com", "*.myapp.com"]
539
)
540
541
@app.middleware("http")
542
async def environment_header(request: Request, call_next):
543
response = await call_next(request)
544
response.headers["X-Environment"] = os.getenv("ENVIRONMENT", "unknown")
545
return response
546
547
@app.get("/")
548
async def read_main():
549
return {"message": "Hello World"}
550
```
551
552
### Custom Middleware Class
553
554
```python
555
from fastapi import FastAPI, Request, Response
556
from starlette.middleware.base import BaseHTTPMiddleware
557
import uuid
558
559
class RequestIDMiddleware(BaseHTTPMiddleware):
560
async def dispatch(self, request: Request, call_next):
561
# Generate unique request ID
562
request_id = str(uuid.uuid4())
563
564
# Add request ID to request state
565
request.state.request_id = request_id
566
567
# Process request
568
response = await call_next(request)
569
570
# Add request ID to response headers
571
response.headers["X-Request-ID"] = request_id
572
573
return response
574
575
app = FastAPI()
576
577
# Add custom middleware class
578
app.add_middleware(RequestIDMiddleware)
579
580
@app.get("/")
581
async def read_main(request: Request):
582
return {
583
"message": "Hello World",
584
"request_id": request.state.request_id
585
}
586
```