pypi-fastapi

1
# Middleware
2

3
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.
4

5
## Capabilities
6

7
### Base Middleware Class
8

9
Base class for creating custom middleware components that process requests and responses.
10

11
```python { .api }
12
class Middleware:
13
    def __init__(self, cls: type, **options: Any) -> None:
14
        """
15
        Base middleware class for custom middleware.
16
        
17
        Parameters:
18
        - cls: Middleware class to instantiate
19
        - options: Configuration options for the middleware
20
        """
21
        self.cls = cls
22
        self.options = options
23
```
24

25
### Middleware Decorator
26

27
Decorator method on FastAPI and APIRouter instances for adding middleware functions.
28

29
```python { .api }
30
def middleware(self, middleware_type: str) -> Callable[[Callable], Callable]:
31
    """
32
    Decorator for adding middleware to the application.
33

34
    Parameters:
35
    - middleware_type: Type of middleware ("http" for HTTP middleware)
36
    
37
    Returns:
38
    Decorator function for middleware registration
39
    """
40
```
41

42
### CORS Middleware
43

44
Cross-Origin Resource Sharing middleware for handling cross-domain requests.
45

46
```python { .api }
47
class CORSMiddleware:
48
    def __init__(
49
        self,
50
        app: ASGIApp,
51
        allow_origins: List[str] = None,
52
        allow_methods: List[str] = None,
53
        allow_headers: List[str] = None,
54
        allow_credentials: bool = False,
55
        allow_origin_regex: str = None,
56
        expose_headers: List[str] = None,
57
        max_age: int = 600
58
    ) -> None:
59
        """
60
        Cross-Origin Resource Sharing middleware.
61
        
62
        Parameters:
63
        - app: ASGI application to wrap
64
        - allow_origins: List of allowed origin URLs
65
        - allow_methods: List of allowed HTTP methods
66
        - allow_headers: List of allowed request headers
67
        - allow_credentials: Allow credentials in cross-origin requests
68
        - allow_origin_regex: Regex pattern for allowed origins
69
        - expose_headers: List of headers to expose to the browser
70
        - max_age: Maximum age for preflight cache
71
        """
72

73
    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
74
        """Process ASGI request with CORS handling."""
75
```
76

77
### GZip Middleware
78

79
Middleware for compressing HTTP responses using GZip compression.
80

81
```python { .api }
82
class GZipMiddleware:
83
    def __init__(
84
        self,
85
        app: ASGIApp,
86
        minimum_size: int = 500,
87
        compresslevel: int = 9
88
    ) -> None:
89
        """
90
        GZip compression middleware.
91
        
92
        Parameters:
93
        - app: ASGI application to wrap
94
        - minimum_size: Minimum response size to compress (bytes)
95
        - compresslevel: GZip compression level (1-9)
96
        """
97

98
    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
99
        """Process ASGI request with GZip compression."""
100
```
101

102
### HTTPS Redirect Middleware
103

104
Middleware for enforcing HTTPS connections by redirecting HTTP requests.
105

106
```python { .api }
107
class HTTPSRedirectMiddleware:
108
    def __init__(self, app: ASGIApp) -> None:
109
        """
110
        HTTPS redirect enforcement middleware.
111
        
112
        Parameters:
113
        - app: ASGI application to wrap
114
        """
115

116
    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
117
        """Process ASGI request with HTTPS redirect."""
118
```
119

120
### Trusted Host Middleware
121

122
Middleware for validating Host headers to prevent Host header attacks.
123

124
```python { .api }
125
class TrustedHostMiddleware:
126
    def __init__(
127
        self,
128
        app: ASGIApp,
129
        allowed_hosts: List[str] = None,
130
        www_redirect: bool = True
131
    ) -> None:
132
        """
133
        Trusted host validation middleware.
134
        
135
        Parameters:
136
        - app: ASGI application to wrap
137
        - allowed_hosts: List of allowed host patterns
138
        - www_redirect: Redirect www subdomain to non-www
139
        """
140

141
    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
142
        """Process ASGI request with host validation."""
143
```
144

145
### WSGI Middleware
146

147
Middleware for mounting WSGI applications within ASGI applications.
148

149
```python { .api }
150
class WSGIMiddleware:
151
    def __init__(self, app: ASGIApp, wsgi_app: WSGIApp) -> None:
152
        """
153
        WSGI application mounting middleware.
154
        
155
        Parameters:
156
        - app: ASGI application to wrap
157
        - wsgi_app: WSGI application to mount
158
        """
159

160
    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
161
        """Process ASGI request with WSGI app mounting."""
162
```
163

164
### Custom Middleware Interface
165

166
Interface for creating custom HTTP middleware functions.
167

168
```python { .api }
169
async def custom_middleware_function(
170
    request: Request,
171
    call_next: Callable[[Request], Awaitable[Response]]
172
) -> Response:
173
    """
174
    Custom middleware function signature.
175
    
176
    Parameters:
177
    - request: HTTP request object
178
    - call_next: Function to call next middleware/route handler
179
    
180
    Returns:
181
    Response object (potentially modified)
182
    """
183
```
184

185
## Usage Examples
186

187
### Basic Custom Middleware
188

189
```python
190
import time
191
from fastapi import FastAPI, Request
192

193
app = FastAPI()
194

195
@app.middleware("http")
196
async def add_process_time_header(request: Request, call_next):
197
    start_time = time.time()
198
    response = await call_next(request)
199
    process_time = time.time() - start_time
200
    response.headers["X-Process-Time"] = str(process_time)
201
    return response
202

203
@app.get("/")
204
async def read_main():
205
    return {"message": "Hello World"}
206
```
207

208
### CORS Middleware Configuration
209

210
```python
211
from fastapi import FastAPI
212
from fastapi.middleware.cors import CORSMiddleware
213

214
app = FastAPI()
215

216
app.add_middleware(
217
    CORSMiddleware,
218
    allow_origins=["http://localhost:3000", "https://myapp.com"],
219
    allow_credentials=True,
220
    allow_methods=["GET", "POST", "PUT", "DELETE"],
221
    allow_headers=["*"],
222
    expose_headers=["X-Custom-Header"],
223
    max_age=600
224
)
225

226
@app.get("/api/data")
227
async def get_data():
228
    return {"data": "This endpoint supports CORS"}
229
```
230

231
### GZip Compression Middleware
232

233
```python
234
from fastapi import FastAPI
235
from fastapi.middleware.gzip import GZipMiddleware
236

237
app = FastAPI()
238

239
# Enable GZip compression for responses larger than 1000 bytes
240
app.add_middleware(GZipMiddleware, minimum_size=1000)
241

242
@app.get("/large-data")
243
async def get_large_data():
244
    # Return large response that will be compressed
245
    return {"data": "x" * 2000, "message": "This response will be compressed"}
246
```
247

248
### HTTPS Redirect Middleware
249

250
```python
251
from fastapi import FastAPI
252
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware
253

254
app = FastAPI()
255

256
# Redirect all HTTP requests to HTTPS
257
app.add_middleware(HTTPSRedirectMiddleware)
258

259
@app.get("/secure-endpoint")
260
async def secure_endpoint():
261
    return {"message": "This endpoint requires HTTPS"}
262
```
263

264
### Trusted Host Middleware
265

266
```python
267
from fastapi import FastAPI
268
from fastapi.middleware.trustedhost import TrustedHostMiddleware
269

270
app = FastAPI()
271

272
# Only allow requests from specific hosts
273
app.add_middleware(
274
    TrustedHostMiddleware,
275
    allowed_hosts=["example.com", "*.example.com", "localhost"]
276
)
277

278
@app.get("/")
279
async def read_main():
280
    return {"message": "Request from trusted host"}
281
```
282

283
### Authentication Middleware
284

285
```python
286
import jwt
287
from fastapi import FastAPI, Request, HTTPException
288
from fastapi.responses import JSONResponse
289

290
app = FastAPI()
291

292
SECRET_KEY = "your-secret-key"
293
ALGORITHM = "HS256"
294

295
@app.middleware("http")
296
async def authenticate_request(request: Request, call_next):
297
    # Skip authentication for certain paths
298
    if request.url.path in ["/login", "/docs", "/openapi.json"]:
299
        response = await call_next(request)
300
        return response
301
    
302
    # Extract token from Authorization header
303
    auth_header = request.headers.get("Authorization")
304
    if not auth_header or not auth_header.startswith("Bearer "):
305
        return JSONResponse(
306
            status_code=401,
307
            content={"error": "Missing or invalid authorization header"}
308
        )
309
    
310
    token = auth_header.replace("Bearer ", "")
311
    
312
    try:
313
        # Verify JWT token
314
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
315
        request.state.user = payload
316
    except jwt.InvalidTokenError:
317
        return JSONResponse(
318
            status_code=401,
319
            content={"error": "Invalid token"}
320
        )
321
    
322
    response = await call_next(request)
323
    return response
324

325
@app.get("/protected")
326
async def protected_route(request: Request):
327
    return {"message": f"Hello {request.state.user['sub']}"}
328
```
329

330
### Logging Middleware
331

332
```python
333
import logging
334
import time
335
from fastapi import FastAPI, Request
336

337
# Configure logging
338
logging.basicConfig(level=logging.INFO)
339
logger = logging.getLogger(__name__)
340

341
app = FastAPI()
342

343
@app.middleware("http")
344
async def log_requests(request: Request, call_next):
345
    start_time = time.time()
346
    
347
    # Log request
348
    logger.info(
349
        f"Request: {request.method} {request.url.path}",
350
        extra={
351
            "method": request.method,
352
            "path": request.url.path,
353
            "query_params": str(request.query_params),
354
            "client": request.client.host if request.client else None
355
        }
356
    )
357
    
358
    # Process request
359
    response = await call_next(request)
360
    
361
    # Log response
362
    process_time = time.time() - start_time
363
    logger.info(
364
        f"Response: {response.status_code} in {process_time:.4f}s",
365
        extra={
366
            "status_code": response.status_code,
367
            "process_time": process_time,
368
            "path": request.url.path
369
        }
370
    )
371
    
372
    return response
373

374
@app.get("/")
375
async def read_main():
376
    return {"message": "Hello World"}
377
```
378

379
### Rate Limiting Middleware
380

381
```python
382
import time
383
from collections import defaultdict
384
from fastapi import FastAPI, Request, HTTPException
385

386
app = FastAPI()
387

388
# Simple in-memory rate limiter
389
rate_limiter = defaultdict(list)
390
RATE_LIMIT = 10  # requests per minute
391
RATE_WINDOW = 60  # seconds
392

393
@app.middleware("http")
394
async def rate_limit_middleware(request: Request, call_next):
395
    client_ip = request.client.host if request.client else "unknown"
396
    current_time = time.time()
397
    
398
    # Clean old requests
399
    rate_limiter[client_ip] = [
400
        req_time for req_time in rate_limiter[client_ip]
401
        if current_time - req_time < RATE_WINDOW
402
    ]
403
    
404
    # Check rate limit
405
    if len(rate_limiter[client_ip]) >= RATE_LIMIT:
406
        raise HTTPException(
407
            status_code=429,
408
            detail="Rate limit exceeded",
409
            headers={"Retry-After": str(RATE_WINDOW)}
410
        )
411
    
412
    # Add current request
413
    rate_limiter[client_ip].append(current_time)
414
    
415
    response = await call_next(request)
416
    return response
417

418
@app.get("/")
419
async def read_main():
420
    return {"message": "Hello World"}
421
```
422

423
### Error Handling Middleware
424

425
```python
426
import traceback
427
from fastapi import FastAPI, Request, HTTPException
428
from fastapi.responses import JSONResponse
429

430
app = FastAPI()
431

432
@app.middleware("http")
433
async def catch_exceptions_middleware(request: Request, call_next):
434
    try:
435
        response = await call_next(request)
436
        return response
437
    except HTTPException:
438
        # Re-raise HTTPExceptions to be handled by FastAPI
439
        raise
440
    except Exception as e:
441
        # Handle unexpected exceptions
442
        error_id = str(hash(str(e) + str(time.time())))
443
        
444
        # Log the full traceback
445
        logger.error(
446
            f"Unhandled exception {error_id}: {str(e)}",
447
            extra={
448
                "error_id": error_id,
449
                "path": request.url.path,
450
                "method": request.method,
451
                "traceback": traceback.format_exc()
452
            }
453
        )
454
        
455
        return JSONResponse(
456
            status_code=500,
457
            content={
458
                "error": "Internal server error",
459
                "error_id": error_id,
460
                "message": "An unexpected error occurred"
461
            }
462
        )
463

464
@app.get("/error")
465
async def trigger_error():
466
    raise ValueError("This is a test error")
467

468
@app.get("/")
469
async def read_main():
470
    return {"message": "Hello World"}
471
```
472

473
### Multiple Middleware Stack
474

475
```python
476
from fastapi import FastAPI, Request
477
from fastapi.middleware.cors import CORSMiddleware
478
from fastapi.middleware.gzip import GZipMiddleware
479
import time
480

481
app = FastAPI()
482

483
# Add multiple middleware in order
484
# Note: Middleware is executed in reverse order of addition
485

486
# 1. GZip (executed last - compresses final response)
487
app.add_middleware(GZipMiddleware, minimum_size=1000)
488

489
# 2. CORS (executed second to last)
490
app.add_middleware(
491
    CORSMiddleware,
492
    allow_origins=["*"],
493
    allow_credentials=True,
494
    allow_methods=["*"],
495
    allow_headers=["*"],
496
)
497

498
# 3. Custom timing middleware (executed first)
499
@app.middleware("http")
500
async def add_timing_header(request: Request, call_next):
501
    start_time = time.time()
502
    response = await call_next(request)
503
    process_time = time.time() - start_time
504
    response.headers["X-Process-Time"] = str(process_time)
505
    return response
506

507
@app.get("/")
508
async def read_main():
509
    return {"message": "Hello World", "data": "x" * 1500}  # Large response for GZip
510
```
511

512
### Conditional Middleware
513

514
```python
515
from fastapi import FastAPI, Request
516
import os
517

518
app = FastAPI()
519

520
# Only add CORS middleware in development
521
if os.getenv("ENVIRONMENT") == "development":
522
    from fastapi.middleware.cors import CORSMiddleware
523
    app.add_middleware(
524
        CORSMiddleware,
525
        allow_origins=["*"],
526
        allow_credentials=True,
527
        allow_methods=["*"],
528
        allow_headers=["*"],
529
    )
530

531
# Security middleware for production
532
if os.getenv("ENVIRONMENT") == "production":
533
    from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware
534
    from fastapi.middleware.trustedhost import TrustedHostMiddleware
535
    
536
    app.add_middleware(HTTPSRedirectMiddleware)
537
    app.add_middleware(
538
        TrustedHostMiddleware,
539
        allowed_hosts=["myapp.com", "*.myapp.com"]
540
    )
541

542
@app.middleware("http")
543
async def environment_header(request: Request, call_next):
544
    response = await call_next(request)
545
    response.headers["X-Environment"] = os.getenv("ENVIRONMENT", "unknown")
546
    return response
547

548
@app.get("/")
549
async def read_main():
550
    return {"message": "Hello World"}
551
```
552

553
### Custom Middleware Class
554

555
```python
556
from fastapi import FastAPI, Request, Response
557
from starlette.middleware.base import BaseHTTPMiddleware
558
import uuid
559

560
class RequestIDMiddleware(BaseHTTPMiddleware):
561
    async def dispatch(self, request: Request, call_next):
562
        # Generate unique request ID
563
        request_id = str(uuid.uuid4())
564
        
565
        # Add request ID to request state
566
        request.state.request_id = request_id
567
        
568
        # Process request
569
        response = await call_next(request)
570
        
571
        # Add request ID to response headers
572
        response.headers["X-Request-ID"] = request_id
573
        
574
        return response
575

576
app = FastAPI()
577

578
# Add custom middleware class
579
app.add_middleware(RequestIDMiddleware)
580

581
@app.get("/")
582
async def read_main(request: Request):
583
    return {
584
        "message": "Hello World",
585
        "request_id": request.state.request_id
586
    }
587
```