0
# Middleware
1

2
Collection of WSGI middleware components for common functionality including static file serving, proxy handling, profiling, and application dispatching. These middleware classes provide reusable functionality that can be wrapped around WSGI applications.
3

4
## Capabilities
5

6
### Shared Data Middleware
7

8
Serves static files during development or for simple deployments without requiring a separate web server.
9

10
```python { .api }
11
class SharedDataMiddleware:
12
    def __init__(self, app, exports, disallow=None, cache=True, cache_timeout=43200, fallback_mimetype="text/plain"):
13
        """
14
        WSGI middleware for serving static files.
15
        
16
        Parameters:
17
        - app: WSGI application to wrap
18
        - exports: Dict mapping URL paths to filesystem paths or package data
19
        - disallow: List of fnmatch patterns for files to deny access to
20
        - cache: Whether to send caching headers
21
        - cache_timeout: Cache timeout in seconds (default: 12 hours)
22
        - fallback_mimetype: Default MIME type for unknown files
23
        """
24
    
25
    def is_allowed(self, filename):
26
        """
27
        Check if a filename is allowed to be served.
28
        
29
        Parameters:
30
        - filename: Filename to check against disallow patterns
31
        
32
        Returns:
33
        True if file is allowed, False if blocked by disallow patterns
34
        """
35
    
36
    def __call__(self, environ, start_response):
37
        """
38
        WSGI application callable.
39
        
40
        Serves static files if path matches exports, otherwise forwards
41
        to the wrapped application.
42
        """
43
```
44

45
### Proxy Fix Middleware
46

47
Handles X-Forwarded headers when the application is behind a proxy server.
48

49
```python { .api }
50
class ProxyFix:
51
    def __init__(self, app, x_for=1, x_proto=1, x_host=0, x_port=0, x_prefix=0):
52
        """
53
        Adjust WSGI environ based on X-Forwarded headers from proxies.
54
        
55
        Parameters:
56
        - app: WSGI application to wrap
57
        - x_for: Number of X-Forwarded-For values to trust (sets REMOTE_ADDR)
58
        - x_proto: Number of X-Forwarded-Proto values to trust (sets wsgi.url_scheme)
59
        - x_host: Number of X-Forwarded-Host values to trust (sets HTTP_HOST, SERVER_NAME, SERVER_PORT)
60
        - x_port: Number of X-Forwarded-Port values to trust (sets HTTP_HOST, SERVER_PORT)
61
        - x_prefix: Number of X-Forwarded-Prefix values to trust (sets SCRIPT_NAME)
62
        
63
        Note: Only trust headers from the number of proxies you expect.
64
        Trusting wrong values is a security vulnerability.
65
        """
66
    
67
    def __call__(self, environ, start_response):
68
        """
69
        WSGI application callable.
70
        
71
        Processes X-Forwarded headers and updates environ accordingly.
72
        Original values are preserved in environ['werkzeug.proxy_fix.orig'].
73
        """
74
```
75

76
### Dispatcher Middleware
77

78
Combines multiple WSGI applications into a single application based on URL paths.
79

80
```python { .api }
81
class DispatcherMiddleware:
82
    def __init__(self, app, mounts=None):
83
        """
84
        Dispatch requests to different applications based on path prefixes.
85
        
86
        Parameters:
87
        - app: Default WSGI application for unmatched paths
88
        - mounts: Dict mapping path prefixes to WSGI applications
89
        """
90
    
91
    def __call__(self, environ, start_response):
92
        """
93
        WSGI application callable.
94
        
95
        Dispatches request to appropriate mounted application or default app.
96
        Updates SCRIPT_NAME and PATH_INFO appropriately for mounted apps.
97
        """
98
```
99

100
### Profiler Middleware
101

102
Profiles each request using Python's cProfile module to identify performance bottlenecks.
103

104
```python { .api }
105
class ProfilerMiddleware:
106
    def __init__(self, app, stream=sys.stdout, sort_by=("time", "calls"), restrictions=(), profile_dir=None, filename_format="{method}.{path}.{elapsed:.0f}ms.{time:.0f}.prof"):
107
        """
108
        Profile WSGI application performance.
109
        
110
        Parameters:
111
        - app: WSGI application to wrap
112
        - stream: Stream to write profiling stats to (None to disable)
113
        - sort_by: Tuple of columns to sort stats by
114
        - restrictions: Tuple of restrictions to filter stats
115
        - profile_dir: Directory to save profile data files
116
        - filename_format: Format string for profile filenames or callable
117
        """
118
    
119
    def __call__(self, environ, start_response):
120
        """
121
        WSGI application callable.
122
        
123
        Profiles the wrapped application and outputs stats according to configuration.
124
        """
125
```
126

127
### Lint Middleware
128

129
Validates WSGI compliance of applications and middleware for debugging.
130

131
```python { .api }
132
class LintMiddleware:
133
    def __init__(self, app):
134
        """
135
        WSGI compliance checking middleware.
136
        
137
        Parameters:
138
        - app: WSGI application to validate
139
        
140
        Note: This middleware checks for WSGI spec compliance and will
141
        raise warnings or errors for violations. Use during development only.
142
        """
143
    
144
    def __call__(self, environ, start_response):
145
        """
146
        WSGI application callable.
147
        
148
        Validates environ dict, start_response callback, and application
149
        return value for WSGI spec compliance.
150
        """
151
```
152

153
### HTTP Proxy Middleware
154

155
Provides HTTP proxy functionality for forwarding requests to remote servers.
156

157
```python { .api }
158
class ProxyMiddleware:
159
    def __init__(self, app, targets, chunk_size=2 << 13, timeout=10):
160
        """
161
        HTTP proxy middleware for forwarding requests.
162
        
163
        Parameters:
164
        - app: WSGI application to wrap (fallback for non-proxied requests)
165
        - targets: Dict mapping path prefixes to target URLs
166
        - chunk_size: Size of chunks when streaming responses
167
        - timeout: Timeout for proxy requests
168
        """
169
    
170
    def __call__(self, environ, start_response):
171
        """
172
        WSGI application callable.
173
        
174
        Forwards requests matching target paths to remote servers,
175
        otherwise passes to wrapped application.
176
        """
177
```
178

179
## Usage Examples
180

181
### Static File Serving
182

183
```python
184
from werkzeug.middleware.shared_data import SharedDataMiddleware
185
from werkzeug.wrappers import Request, Response
186
import os
187

188
def app(environ, start_response):
189
    request = Request(environ)
190
    response = Response(f'API endpoint: {request.path}')
191
    return response(environ, start_response)
192

193
# Serve static files from multiple locations
194
wrapped_app = SharedDataMiddleware(app, {
195
    '/static': os.path.join(os.path.dirname(__file__), 'static'),
196
    '/uploads': '/var/www/uploads',
197
    '/assets': ('mypackage', 'assets'),  # Package data
198
})
199

200
# With custom configuration
201
wrapped_app = SharedDataMiddleware(
202
    app,
203
    {'/static': './static'},
204
    disallow=['*.secret', '*.key', '.htaccess'],  # Block sensitive files
205
    cache=True,
206
    cache_timeout=3600,  # 1 hour cache
207
    fallback_mimetype='application/octet-stream'
208
)
209

210
if __name__ == '__main__':
211
    from werkzeug.serving import run_simple
212
    run_simple('localhost', 8000, wrapped_app, use_reloader=True)
213
```
214

215
### Proxy Configuration
216

217
```python
218
from werkzeug.middleware.proxy_fix import ProxyFix
219

220
def app(environ, start_response):
221
    # Access real client information after proxy processing
222
    remote_addr = environ.get('REMOTE_ADDR')
223
    scheme = environ.get('wsgi.url_scheme')
224
    host = environ.get('HTTP_HOST')
225
    
226
    response = Response(f'Client: {remote_addr}, Scheme: {scheme}, Host: {host}')
227
    return response(environ, start_response)
228

229
# Configure for deployment behind nginx
230
# nginx sets X-Forwarded-For and X-Forwarded-Proto
231
proxy_app = ProxyFix(app, x_for=1, x_proto=1)
232

233
# Configure for deployment behind multiple proxies
234
# Load balancer -> nginx -> app
235
multi_proxy_app = ProxyFix(app, x_for=2, x_proto=1, x_host=1)
236

237
# Check original values if needed
238
def debug_app(environ, start_response):
239
    orig = environ.get('werkzeug.proxy_fix.orig', {})
240
    current_addr = environ.get('REMOTE_ADDR')
241
    original_addr = orig.get('REMOTE_ADDR')
242
    
243
    response = Response(f'Current: {current_addr}, Original: {original_addr}')
244
    return response(environ, start_response)
245

246
if __name__ == '__main__':
247
    from werkzeug.serving import run_simple
248
    run_simple('localhost', 8000, proxy_app)
249
```
250

251
### Application Dispatching
252

253
```python
254
from werkzeug.middleware.dispatcher import DispatcherMiddleware
255
from werkzeug.wrappers import Request, Response
256

257
# API application
258
def api_app(environ, start_response):
259
    request = Request(environ)
260
    
261
    if request.path == '/users':
262
        response = Response('{"users": []}', mimetype='application/json')
263
    else:
264
        response = Response('{"error": "Not found"}', status=404)
265
    
266
    return response(environ, start_response)
267

268
# Admin application
269
def admin_app(environ, start_response):
270
    request = Request(environ)
271
    response = Response(f'<h1>Admin Panel</h1><p>Path: {request.path}</p>', 
272
                       mimetype='text/html')
273
    return response(environ, start_response)
274

275
# Frontend application (catch-all)
276
def frontend_app(environ, start_response):
277
    request = Request(environ)
278
    
279
    # Serve SPA for all frontend routes
280
    if request.path.startswith('/app/'):
281
        html = '''
282
        <html>
283
        <head><title>My App</title></head>
284
        <body>
285
            <div id="app">Single Page Application</div>
286
            <script>console.log("Route: " + location.pathname)</script>
287
        </body>
288
        </html>
289
        '''
290
        response = Response(html, mimetype='text/html')
291
    else:
292
        response = Response('Welcome! Try /app/, /api/, or /admin/')
293
    
294
    return response(environ, start_response)
295

296
# Combine applications
297
combined_app = DispatcherMiddleware(frontend_app, {
298
    '/api': api_app,
299
    '/admin': admin_app,
300
})
301

302
if __name__ == '__main__':
303
    from werkzeug.serving import run_simple
304
    run_simple('localhost', 8000, combined_app, use_reloader=True)
305
```
306

307
### Performance Profiling
308

309
```python
310
from werkzeug.middleware.profiler import ProfilerMiddleware
311
from werkzeug.wrappers import Request, Response
312
import time
313
import random
314

315
def slow_app(environ, start_response):
316
    request = Request(environ)
317
    
318
    # Simulate different performance characteristics
319
    if request.path == '/slow':
320
        time.sleep(0.5)  # Simulate slow operation
321
        
322
    elif request.path == '/compute':
323
        # CPU intensive operation
324
        result = sum(i**2 for i in range(10000))
325
        
326
    elif request.path == '/random':
327
        # Variable performance
328
        time.sleep(random.uniform(0.1, 0.3))
329
    
330
    response = Response(f'Processed {request.path}')
331
    return response(environ, start_response)
332

333
# Profile to stdout with custom sorting
334
profiled_app = ProfilerMiddleware(
335
    slow_app,
336
    sort_by=('cumulative', 'calls'),
337
    restrictions=(30,)  # Show top 30 functions
338
)
339

340
# Save profile data to files
341
import os
342
profile_dir = './profiles'
343
os.makedirs(profile_dir, exist_ok=True)
344

345
file_profiled_app = ProfilerMiddleware(
346
    slow_app,
347
    stream=None,  # Don't print to stdout
348
    profile_dir=profile_dir,
349
    filename_format='{method}-{path}-{elapsed:.0f}ms.prof'
350
)
351

352
# Custom filename generator
353
def custom_filename(environ):
354
    profiler_info = environ['werkzeug.profiler']
355
    path = environ.get('PATH_INFO', 'root').replace('/', '-')
356
    elapsed = profiler_info['elapsed']
357
    return f'profile-{path}-{elapsed:.1f}ms.prof'
358

359
custom_profiled_app = ProfilerMiddleware(
360
    slow_app,
361
    profile_dir=profile_dir,
362
    filename_format=custom_filename
363
)
364

365
if __name__ == '__main__':
366
    from werkzeug.serving import run_simple
367
    print("Visit /slow, /compute, or /random to see profiling output")
368
    run_simple('localhost', 8000, profiled_app, use_reloader=True)
369
```
370

371
### WSGI Compliance Testing
372

373
```python
374
from werkzeug.middleware.lint import LintMiddleware
375
from werkzeug.wrappers import Response
376

377
# Application with potential WSGI violations
378
def problematic_app(environ, start_response):
379
    # This app has some WSGI compliance issues
380
    
381
    # Issue 1: Not checking if start_response was called
382
    response = Response('Hello')
383
    
384
    # Issue 2: Returning string instead of bytes iterable
385
    # return 'This would cause a lint error'  # Wrong!
386
    
387
    # Correct WSGI return
388
    return response(environ, start_response)
389

390
# Wrap with lint middleware for development
391
linted_app = LintMiddleware(problematic_app)
392

393
# This will catch and warn about WSGI spec violations
394
if __name__ == '__main__':
395
    from werkzeug.serving import run_simple
396
    run_simple('localhost', 8000, linted_app, use_reloader=True)
397
```
398

399
### HTTP Proxy
400

401
```python
402
from werkzeug.middleware.http_proxy import ProxyMiddleware
403

404
def local_app(environ, start_response):
405
    request = Request(environ)
406
    response = Response(f'Local app handling: {request.path}')
407
    return response(environ, start_response)
408

409
# Proxy certain paths to remote services
410
proxied_app = ProxyMiddleware(local_app, {
411
    '/api/external': 'https://api.example.com',
412
    '/cdn': 'https://cdn.example.com',
413
})
414

415
if __name__ == '__main__':
416
    from werkzeug.serving import run_simple
417
    # Requests to /api/external/* will be forwarded to api.example.com
418
    # Other requests handled by local_app
419
    run_simple('localhost', 8000, proxied_app)
420
```
421

422
### Combining Multiple Middleware
423

424
```python
425
from werkzeug.middleware.shared_data import SharedDataMiddleware
426
from werkzeug.middleware.proxy_fix import ProxyFix
427
from werkzeug.middleware.profiler import ProfilerMiddleware
428
from werkzeug.middleware.dispatcher import DispatcherMiddleware
429

430
def api_app(environ, start_response):
431
    response = Response('{"api": "v1"}', mimetype='application/json')
432
    return response(environ, start_response)
433

434
def web_app(environ, start_response):
435
    response = Response('<h1>Web Application</h1>', mimetype='text/html')
436
    return response(environ, start_response)
437

438
# Layer middleware (applied inside-out)
439
app = DispatcherMiddleware(web_app, {'/api': api_app})
440

441
# Add static file serving
442
app = SharedDataMiddleware(app, {
443
    '/static': './static',
444
    '/favicon.ico': './static/favicon.ico'
445
})
446

447
# Add proxy fix for production deployment
448
app = ProxyFix(app, x_for=1, x_proto=1)
449

450
# Add profiling for development
451
if __name__ == '__main__':
452
    # Only add profiler in development
453
    app = ProfilerMiddleware(app, sort_by=('cumulative',))
454
    
455
    from werkzeug.serving import run_simple
456
    run_simple('localhost', 8000, app, use_reloader=True)
457
```
458

459
### Development vs Production Configuration
460

461
```python
462
import os
463
from werkzeug.middleware.shared_data import SharedDataMiddleware
464
from werkzeug.middleware.proxy_fix import ProxyFix
465
from werkzeug.middleware.profiler import ProfilerMiddleware
466

467
def create_app():
468
    def app(environ, start_response):
469
        response = Response('Hello World!')
470
        return response(environ, start_response)
471
    
472
    return app
473

474
def wrap_app_for_environment(app):
475
    environment = os.getenv('ENVIRONMENT', 'development')
476
    
477
    if environment == 'development':
478
        # Development middleware
479
        app = SharedDataMiddleware(app, {'/static': './static'})
480
        app = ProfilerMiddleware(app)
481
        
482
    elif environment == 'production':
483
        # Production middleware  
484
        app = ProxyFix(app, x_for=1, x_proto=1, x_host=1)
485
        # Note: Use proper web server for static files in production
486
        
487
    return app
488

489
if __name__ == '__main__':
490
    from werkzeug.serving import run_simple
491
    
492
    app = create_app()
493
    app = wrap_app_for_environment(app)
494
    
495
    run_simple('localhost', 8000, app, use_reloader=True)
496
```
497

498
### Custom Middleware Example
499

500
```python
501
from werkzeug.wrappers import Request, Response
502

503
class RequestTimingMiddleware:
504
    """Custom middleware to add request timing headers."""
505
    
506
    def __init__(self, app):
507
        self.app = app
508
    
509
    def __call__(self, environ, start_response):
510
        import time
511
        
512
        start_time = time.time()
513
        
514
        def timing_start_response(status, headers, exc_info=None):
515
            # Add timing header
516
            elapsed = time.time() - start_time
517
            headers.append(('X-Response-Time', f'{elapsed:.3f}s'))
518
            return start_response(status, headers, exc_info)
519
        
520
        return self.app(environ, timing_start_response)
521

522
def app(environ, start_response):
523
    import time
524
    time.sleep(0.1)  # Simulate processing
525
    response = Response('Timed response')
526
    return response(environ, start_response)
527

528
# Apply custom middleware
529
timed_app = RequestTimingMiddleware(app)
530

531
if __name__ == '__main__':
532
    from werkzeug.serving import run_simple
533
    run_simple('localhost', 8000, timed_app)
534
```