0
# Middleware and Hooks
1
2
Middleware system for implementing cross-cutting concerns like authentication, CORS, and logging, combined with hook decorators for before/after request processing in Falcon applications.
3
4
## Capabilities
5
6
### Middleware System
7
8
Pluggable middleware components that process requests and responses across all routes.
9
10
```python { .api }
11
# Middleware interface (informal protocol)
12
class MiddlewareComponent:
13
def process_request(self, req: Request, resp: Response):
14
"""
15
Process request before routing to resource.
16
17
Args:
18
req: Request object
19
resp: Response object (can be modified)
20
"""
21
22
def process_resource(self, req: Request, resp: Response, resource: object, params: dict):
23
"""
24
Process request after routing but before calling resource method.
25
26
Args:
27
req: Request object
28
resp: Response object
29
resource: Target resource object
30
params: Route parameters
31
"""
32
33
def process_response(self, req: Request, resp: Response, resource: object, req_succeeded: bool):
34
"""
35
Process response after resource method execution.
36
37
Args:
38
req: Request object
39
resp: Response object
40
resource: Resource object that handled the request
41
req_succeeded: Whether request processing succeeded
42
"""
43
```
44
45
#### Middleware Registration
46
47
```python
48
import falcon
49
50
class AuthMiddleware:
51
def process_request(self, req, resp):
52
# Check authentication
53
token = req.get_header('Authorization')
54
if not token:
55
raise falcon.HTTPUnauthorized(
56
title='Authentication required',
57
description='Please provide valid authentication token'
58
)
59
60
# Validate token and set user context
61
user = validate_token(token)
62
req.context.user = user
63
64
class LoggingMiddleware:
65
def process_request(self, req, resp):
66
req.context.start_time = time.time()
67
print(f"Request: {req.method} {req.path}")
68
69
def process_response(self, req, resp, resource, req_succeeded):
70
duration = time.time() - req.context.start_time
71
print(f"Response: {resp.status} ({duration:.3f}s)")
72
73
# Register middleware
74
app = falcon.App(middleware=[
75
AuthMiddleware(),
76
LoggingMiddleware()
77
])
78
79
# Or add middleware after app creation
80
app.add_middleware(AuthMiddleware())
81
```
82
83
### Built-in CORS Middleware
84
85
Comprehensive CORS (Cross-Origin Resource Sharing) middleware with configurable policies.
86
87
```python { .api }
88
class CORSMiddleware:
89
def __init__(
90
self,
91
allow_origins: str | list = '*',
92
expose_headers: list = None,
93
allow_credentials: bool = None,
94
allow_private_network: bool = False,
95
allow_methods: list = None,
96
allow_headers: list = None,
97
max_age: int = None
98
):
99
"""
100
CORS middleware for handling cross-origin requests.
101
102
Args:
103
allow_origins: Allowed origins ('*' or list of origins)
104
expose_headers: Headers to expose to client
105
allow_credentials: Allow credentials in CORS requests
106
allow_private_network: Allow private network access
107
allow_methods: Allowed HTTP methods (defaults to all)
108
allow_headers: Allowed request headers
109
max_age: Preflight cache time in seconds
110
"""
111
112
def process_request(self, req: Request, resp: Response):
113
"""Process CORS preflight requests"""
114
115
def process_response(self, req: Request, resp: Response, resource: object, req_succeeded: bool):
116
"""Add CORS headers to responses"""
117
```
118
119
#### CORS Middleware Usage
120
121
```python
122
import falcon
123
124
# Basic CORS setup
125
cors = falcon.CORSMiddleware(
126
allow_origins=['https://example.com', 'https://app.example.com'],
127
allow_credentials=True
128
)
129
130
app = falcon.App(middleware=[cors])
131
132
# Or enable CORS with app creation
133
app = falcon.App(cors_enable=True) # Uses default CORS settings
134
135
# Advanced CORS configuration
136
cors = falcon.CORSMiddleware(
137
allow_origins='*',
138
expose_headers=['X-Custom-Header'],
139
allow_credentials=False,
140
allow_methods=['GET', 'POST', 'PUT', 'DELETE'],
141
allow_headers=['Content-Type', 'Authorization'],
142
max_age=86400 # 24 hours
143
)
144
```
145
146
### Hook Decorators
147
148
Decorators for executing functions before and after resource method execution.
149
150
```python { .api }
151
def before(action: callable, *args, **kwargs) -> callable:
152
"""
153
Execute function before responder method.
154
155
Args:
156
action: Function to execute before responder
157
*args: Arguments to pass to action function
158
**kwargs: Keyword arguments to pass to action function
159
160
Returns:
161
Decorator function
162
"""
163
164
def after(action: callable, *args, **kwargs) -> callable:
165
"""
166
Execute function after responder method.
167
168
Args:
169
action: Function to execute after responder
170
*args: Arguments to pass to action function
171
**kwargs: Keyword arguments to pass to action function
172
173
Returns:
174
Decorator function
175
"""
176
177
# Hook action function signature
178
def hook_action(req: Request, resp: Response, resource: object, params: dict, *args, **kwargs):
179
"""
180
Hook action function signature.
181
182
Args:
183
req: Request object
184
resp: Response object
185
resource: Resource object being processed
186
params: Route parameters
187
*args: Additional positional arguments
188
**kwargs: Additional keyword arguments
189
"""
190
```
191
192
#### Hook Usage Examples
193
194
```python
195
import falcon
196
from falcon import before, after
197
198
def validate_user_input(req, resp, resource, params):
199
"""Validate request data before processing"""
200
if req.content_length and req.content_length > 1024 * 1024: # 1MB limit
201
raise falcon.HTTPContentTooLarge(
202
title='Request too large',
203
description='Request body cannot exceed 1MB'
204
)
205
206
def log_user_action(req, resp, resource, params, action_name):
207
"""Log user actions after processing"""
208
user_id = req.context.get('user_id')
209
if user_id:
210
log_action(user_id, action_name, params)
211
212
def cache_response(req, resp, resource, params, cache_time=300):
213
"""Set cache headers after processing"""
214
if resp.status.startswith('2'): # Success responses only
215
resp.set_header('Cache-Control', f'max-age={cache_time}')
216
217
class UserResource:
218
@before(validate_user_input)
219
@after(log_user_action, 'user_create')
220
@after(cache_response, cache_time=600)
221
def on_post(self, req, resp):
222
"""Create new user with validation and logging"""
223
user_data = req.media
224
new_user = create_user(user_data)
225
resp.status = falcon.HTTP_201
226
resp.media = new_user
227
228
@after(log_user_action, 'user_view')
229
def on_get(self, req, resp, user_id):
230
"""Get user with action logging"""
231
user = get_user(user_id)
232
if not user:
233
raise falcon.HTTPNotFound()
234
resp.media = user
235
```
236
237
### Custom Middleware Examples
238
239
Common middleware implementation patterns for various use cases.
240
241
#### Authentication Middleware
242
243
```python
244
import jwt
245
import falcon
246
247
class JWTMiddleware:
248
def __init__(self, secret_key, algorithm='HS256', exempt_routes=None):
249
self.secret_key = secret_key
250
self.algorithm = algorithm
251
self.exempt_routes = exempt_routes or []
252
253
def process_request(self, req, resp):
254
# Skip authentication for exempt routes
255
if req.path in self.exempt_routes:
256
return
257
258
# Extract JWT token
259
auth_header = req.get_header('Authorization')
260
if not auth_header or not auth_header.startswith('Bearer '):
261
raise falcon.HTTPUnauthorized(
262
title='Missing authentication',
263
description='Bearer token required in Authorization header'
264
)
265
266
token = auth_header[7:] # Remove 'Bearer ' prefix
267
268
try:
269
# Decode and validate JWT
270
payload = jwt.decode(token, self.secret_key, algorithms=[self.algorithm])
271
req.context.user = payload
272
except jwt.ExpiredSignatureError:
273
raise falcon.HTTPUnauthorized(
274
title='Token expired',
275
description='Authentication token has expired'
276
)
277
except jwt.InvalidTokenError:
278
raise falcon.HTTPUnauthorized(
279
title='Invalid token',
280
description='Authentication token is invalid'
281
)
282
283
# Usage
284
jwt_middleware = JWTMiddleware(
285
secret_key='your-secret-key',
286
exempt_routes=['/health', '/login', '/register']
287
)
288
app = falcon.App(middleware=[jwt_middleware])
289
```
290
291
#### Request/Response Logging Middleware
292
293
```python
294
import json
295
import logging
296
import time
297
import falcon
298
299
class DetailedLoggingMiddleware:
300
def __init__(self, logger=None, log_body=False):
301
self.logger = logger or logging.getLogger(__name__)
302
self.log_body = log_body
303
304
def process_request(self, req, resp):
305
req.context.start_time = time.time()
306
307
# Log request details
308
self.logger.info(f"Request: {req.method} {req.path}")
309
self.logger.debug(f"Headers: {dict(req.headers)}")
310
self.logger.debug(f"Query params: {req.params}")
311
312
if self.log_body and req.content_length:
313
try:
314
# Log request body (be careful with sensitive data)
315
body = req.bounded_stream.read()
316
req.bounded_stream = io.BytesIO(body) # Reset stream
317
self.logger.debug(f"Request body: {body.decode('utf-8')[:1000]}")
318
except Exception as e:
319
self.logger.warning(f"Could not log request body: {e}")
320
321
def process_response(self, req, resp, resource, req_succeeded):
322
duration = time.time() - req.context.start_time
323
324
# Log response details
325
self.logger.info(f"Response: {resp.status} ({duration:.3f}s)")
326
327
if not req_succeeded:
328
self.logger.error(f"Request failed for {req.method} {req.path}")
329
330
# Usage
331
logging_middleware = DetailedLoggingMiddleware(
332
logger=logging.getLogger('falcon.requests'),
333
log_body=True
334
)
335
```
336
337
#### Rate Limiting Middleware
338
339
```python
340
import time
341
from collections import defaultdict, deque
342
import falcon
343
344
class RateLimitMiddleware:
345
def __init__(self, calls_per_minute=100, per_ip=True):
346
self.calls_per_minute = calls_per_minute
347
self.per_ip = per_ip
348
self.call_history = defaultdict(deque)
349
350
def _get_client_id(self, req):
351
"""Get client identifier for rate limiting"""
352
if self.per_ip:
353
# Use real IP if behind proxy
354
return req.get_header('X-Forwarded-For') or req.env.get('REMOTE_ADDR')
355
else:
356
# Use user ID from context if authenticated
357
return getattr(req.context, 'user_id', 'anonymous')
358
359
def process_request(self, req, resp):
360
client_id = self._get_client_id(req)
361
now = time.time()
362
363
# Clean old entries (older than 1 minute)
364
history = self.call_history[client_id]
365
while history and history[0] < now - 60:
366
history.popleft()
367
368
# Check rate limit
369
if len(history) >= self.calls_per_minute:
370
raise falcon.HTTPTooManyRequests(
371
title='Rate limit exceeded',
372
description=f'Maximum {self.calls_per_minute} calls per minute allowed'
373
)
374
375
# Record this call
376
history.append(now)
377
378
# Usage
379
rate_limiter = RateLimitMiddleware(calls_per_minute=60, per_ip=True)
380
```
381
382
### Async Middleware
383
384
Middleware support for ASGI applications with async processing.
385
386
```python { .api }
387
# Async middleware interface
388
class AsyncMiddlewareComponent:
389
async def process_request(self, req: Request, resp: Response):
390
"""Async request processing"""
391
392
async def process_resource(self, req: Request, resp: Response, resource: object, params: dict):
393
"""Async resource processing"""
394
395
async def process_response(self, req: Request, resp: Response, resource: object, req_succeeded: bool):
396
"""Async response processing"""
397
```
398
399
#### Async Middleware Example
400
401
```python
402
import falcon.asgi
403
import asyncio
404
import aioredis
405
406
class AsyncCacheMiddleware:
407
def __init__(self, redis_url='redis://localhost'):
408
self.redis = None
409
410
async def process_request(self, req, resp):
411
# Initialize Redis connection if needed
412
if not self.redis:
413
self.redis = await aioredis.from_url('redis://localhost')
414
415
# Check cache for GET requests
416
if req.method == 'GET':
417
cache_key = f"cache:{req.path}:{req.query_string}"
418
cached_response = await self.redis.get(cache_key)
419
420
if cached_response:
421
resp.media = json.loads(cached_response)
422
resp.complete = True # Skip further processing
423
424
async def process_response(self, req, resp, resource, req_succeeded):
425
# Cache successful GET responses
426
if (req.method == 'GET' and req_succeeded and
427
resp.status.startswith('2') and hasattr(resp, 'media')):
428
429
cache_key = f"cache:{req.path}:{req.query_string}"
430
await self.redis.setex(
431
cache_key,
432
300, # 5 minutes TTL
433
json.dumps(resp.media)
434
)
435
436
# Usage with ASGI app
437
app = falcon.asgi.App(middleware=[AsyncCacheMiddleware()])
438
```
439
440
### Error Handling in Middleware
441
442
Best practices for error handling within middleware components.
443
444
```python
445
class SafeMiddleware:
446
def process_request(self, req, resp):
447
try:
448
# Middleware logic here
449
self._process_request_logic(req, resp)
450
except Exception as e:
451
# Log error and optionally convert to HTTP error
452
logger.error(f"Middleware error: {e}")
453
454
# Re-raise as HTTP error for proper handling
455
raise falcon.HTTPInternalServerError(
456
title='Middleware processing failed',
457
description='An internal error occurred while processing the request'
458
)
459
460
def _process_request_logic(self, req, resp):
461
"""Actual middleware logic"""
462
pass
463
```
464
465
## Types
466
467
```python { .api }
468
# Middleware system (informal protocols - no concrete types)
469
MiddlewareComponent: protocol # Middleware interface pattern
470
471
# Built-in middleware
472
CORSMiddleware: type
473
474
# Hook decorators
475
before: callable # Before hook decorator
476
after: callable # After hook decorator
477
478
# Hook action signature (function type)
479
HookAction: callable # (req, resp, resource, params, *args, **kwargs) -> None
480
```