0
# Helper Functions and Utilities
1
2
HTTP helpers, URL generation, file serving, flash messaging, JSON support, and advanced features for comprehensive web development workflows.
3
4
## Capabilities
5
6
### HTTP Response Helpers
7
8
Core HTTP utilities for response creation, error handling, and request flow control.
9
10
```python { .api }
11
def abort(code: int, *args, **kwargs) -> NoReturn:
12
"""
13
Raise HTTP exception with specified status code.
14
15
Args:
16
code: HTTP status code (400, 404, 500, etc.)
17
*args: Additional arguments for exception
18
**kwargs: Additional keyword arguments
19
20
Raises:
21
HTTPException: Always raises, never returns
22
"""
23
24
async def make_response(*args) -> Response:
25
"""
26
Create response object from various input types.
27
28
Args:
29
*args: Response data (string, dict, tuple, Response, etc.)
30
31
Returns:
32
Response object
33
"""
34
35
def redirect(location: str, code: int = 302, Response=None) -> Response:
36
"""
37
Create redirect response.
38
39
Args:
40
location: URL to redirect to
41
code: HTTP redirect status code (301, 302, 303, 307, 308)
42
Response: Custom response class (optional)
43
44
Returns:
45
Redirect response object
46
"""
47
```
48
49
### URL Generation
50
51
URL construction and endpoint resolution with support for external URLs and query parameters.
52
53
```python { .api }
54
def url_for(endpoint: str, **values) -> str:
55
"""
56
Generate URL for endpoint with given parameters.
57
58
Args:
59
endpoint: Endpoint name (function name or 'blueprint.function_name')
60
**values: URL parameters and query string arguments
61
62
Returns:
63
Generated URL string
64
65
Example:
66
url_for('user_profile', username='john', page=2)
67
# Returns: '/user/john?page=2'
68
"""
69
```
70
71
### File Serving
72
73
Async file serving with MIME type detection, caching headers, and download attachment support.
74
75
```python { .api }
76
async def send_file(
77
filename_or_io,
78
mimetype: str | None = None,
79
as_attachment: bool = False,
80
download_name: str | None = None,
81
conditional: bool = True,
82
etag: bool | str = True,
83
last_modified: datetime | None = None,
84
max_age: int | None = None,
85
cache_timeout: int | None = None
86
) -> Response:
87
"""
88
Send file as response with proper headers.
89
90
Args:
91
filename_or_io: File path or file-like object
92
mimetype: MIME type (auto-detected if None)
93
as_attachment: Force download vs inline display
94
download_name: Filename for download
95
conditional: Enable conditional requests (304 responses)
96
etag: ETag value or True for auto-generation
97
last_modified: Last modified datetime
98
max_age: Cache max-age in seconds
99
cache_timeout: Deprecated, use max_age
100
101
Returns:
102
File response with appropriate headers
103
"""
104
105
async def send_from_directory(
106
directory: str,
107
filename: str,
108
**kwargs
109
) -> Response:
110
"""
111
Send file from directory with path safety checks.
112
113
Args:
114
directory: Base directory path
115
filename: Filename within directory (path traversal protected)
116
**kwargs: Additional arguments for send_file()
117
118
Returns:
119
File response
120
121
Raises:
122
NotFound: If file doesn't exist or path is invalid
123
"""
124
```
125
126
### Flash Messaging
127
128
Session-based flash messaging system for user notifications across requests.
129
130
```python { .api }
131
async def flash(message: str, category: str = "message"):
132
"""
133
Add flash message to session for next request.
134
135
Args:
136
message: Message text to display
137
category: Message category ('message', 'error', 'warning', 'info')
138
"""
139
140
def get_flashed_messages(
141
with_categories: bool = False,
142
category_filter: tuple = ()
143
) -> list:
144
"""
145
Retrieve and clear flash messages from session.
146
147
Args:
148
with_categories: Return (category, message) tuples instead of just messages
149
category_filter: Only return messages from specified categories
150
151
Returns:
152
List of messages or (category, message) tuples
153
"""
154
```
155
156
### JSON Support
157
158
JSON response creation and data serialization with customizable encoding.
159
160
```python { .api }
161
def jsonify(*args, **kwargs) -> Response:
162
"""
163
Create JSON response from data.
164
165
Args:
166
*args: Positional arguments (single value becomes response)
167
**kwargs: Keyword arguments (become JSON object)
168
169
Returns:
170
Response with JSON content and appropriate headers
171
172
Examples:
173
jsonify({'key': 'value'}) # JSON object
174
jsonify([1, 2, 3]) # JSON array
175
jsonify(key='value') # JSON object from kwargs
176
"""
177
178
def dumps(obj, **kwargs) -> str:
179
"""
180
Serialize object to JSON string using app's JSON encoder.
181
182
Args:
183
obj: Object to serialize
184
**kwargs: Additional arguments for json.dumps()
185
186
Returns:
187
JSON string
188
"""
189
190
def dump(obj, fp, **kwargs):
191
"""
192
Serialize object to JSON file using app's JSON encoder.
193
194
Args:
195
obj: Object to serialize
196
fp: File-like object to write to
197
**kwargs: Additional arguments for json.dump()
198
"""
199
200
def loads(s: str, **kwargs):
201
"""
202
Deserialize JSON string using app's JSON decoder.
203
204
Args:
205
s: JSON string to deserialize
206
**kwargs: Additional arguments for json.loads()
207
208
Returns:
209
Deserialized Python object
210
"""
211
212
def load(fp, **kwargs):
213
"""
214
Deserialize JSON file using app's JSON decoder.
215
216
Args:
217
fp: File-like object to read from
218
**kwargs: Additional arguments for json.load()
219
220
Returns:
221
Deserialized Python object
222
"""
223
```
224
225
### Advanced Features
226
227
Specialized features for HTTP/2 push promises, streaming with context, and template integration.
228
229
```python { .api }
230
async def make_push_promise(path: str):
231
"""
232
Create HTTP/2 server push promise.
233
234
Args:
235
path: Resource path to push to client
236
237
Note:
238
Only works with HTTP/2 compatible servers
239
"""
240
241
def stream_with_context(func: Callable):
242
"""
243
Decorator to stream response while preserving request context.
244
245
Args:
246
func: Generator function that yields response chunks
247
248
Returns:
249
Decorated function that maintains context during streaming
250
"""
251
252
def get_template_attribute(template_name: str, attribute: str):
253
"""
254
Get attribute from template without rendering.
255
256
Args:
257
template_name: Template file name
258
attribute: Attribute name to retrieve
259
260
Returns:
261
Template attribute value
262
"""
263
```
264
265
### Usage Examples
266
267
#### HTTP Response Helpers
268
269
```python
270
from quart import Quart, abort, make_response, redirect, url_for
271
272
app = Quart(__name__)
273
274
@app.route('/user/<int:user_id>')
275
async def user_profile(user_id):
276
user = await get_user(user_id)
277
if not user:
278
abort(404) # Raises 404 Not Found
279
280
return f"User: {user.name}"
281
282
@app.route('/admin')
283
async def admin_area():
284
if not current_user.is_admin:
285
abort(403) # Raises 403 Forbidden
286
287
return "Admin Dashboard"
288
289
@app.route('/custom-response')
290
async def custom_response():
291
# Create custom response
292
response = await make_response("Custom content")
293
response.status_code = 201
294
response.headers['X-Custom'] = 'value'
295
return response
296
297
@app.route('/login', methods=['POST'])
298
async def login():
299
# Process login...
300
if login_successful:
301
return redirect(url_for('dashboard'))
302
else:
303
return redirect(url_for('login_form', error='invalid'))
304
305
@app.route('/old-url')
306
async def old_url():
307
# Permanent redirect
308
return redirect(url_for('new_url'), code=301)
309
```
310
311
#### File Serving
312
313
```python
314
from quart import Quart, send_file, send_from_directory
315
316
app = Quart(__name__)
317
318
@app.route('/download/<filename>')
319
async def download_file(filename):
320
# Send file as download
321
return await send_from_directory(
322
'uploads',
323
filename,
324
as_attachment=True,
325
download_name=f"document_{filename}"
326
)
327
328
@app.route('/image/<image_id>')
329
async def serve_image(image_id):
330
# Serve image with caching
331
image_path = f"/images/{image_id}.jpg"
332
return await send_file(
333
image_path,
334
mimetype='image/jpeg',
335
max_age=3600 # Cache for 1 hour
336
)
337
338
@app.route('/report/pdf')
339
async def generate_report():
340
# Generate PDF and send
341
pdf_data = await generate_pdf_report()
342
343
return await send_file(
344
io.BytesIO(pdf_data),
345
mimetype='application/pdf',
346
as_attachment=True,
347
download_name=f'report_{datetime.now().date()}.pdf'
348
)
349
350
@app.route('/avatar/<username>')
351
async def user_avatar(username):
352
# Serve user avatar with fallback
353
avatar_path = f"/avatars/{username}.png"
354
355
if os.path.exists(avatar_path):
356
return await send_file(avatar_path, max_age=1800)
357
else:
358
# Send default avatar
359
return await send_file("/static/default_avatar.png")
360
```
361
362
#### Flash Messaging
363
364
```python
365
from quart import Quart, flash, get_flashed_messages, render_template, redirect, url_for
366
367
app = Quart(__name__)
368
app.secret_key = 'secret-key-for-sessions'
369
370
@app.route('/form', methods=['POST'])
371
async def process_form():
372
form_data = await request.form
373
374
try:
375
# Process form data
376
result = await process_user_data(form_data)
377
378
await flash(f'Successfully processed {result.count} items!', 'success')
379
await flash('Data has been saved to the database.', 'info')
380
381
except ValidationError as e:
382
await flash(f'Validation error: {e}', 'error')
383
except Exception as e:
384
await flash('An unexpected error occurred.', 'error')
385
386
return redirect(url_for('show_form'))
387
388
@app.route('/form')
389
async def show_form():
390
# Get flash messages for template
391
messages = get_flashed_messages(with_categories=True)
392
393
return await render_template('form.html', messages=messages)
394
395
@app.route('/admin/bulk-update', methods=['POST'])
396
async def bulk_update():
397
try:
398
updates = await perform_bulk_update()
399
400
for category, count in updates.items():
401
await flash(f'Updated {count} {category} records', 'success')
402
403
except Exception as e:
404
await flash('Bulk update failed', 'error')
405
await flash(str(e), 'error')
406
407
return redirect(url_for('admin_dashboard'))
408
409
# Template usage (form.html):
410
"""
411
{% for category, message in messages %}
412
<div class="alert alert-{{ category }}">{{ message }}</div>
413
{% endfor %}
414
"""
415
```
416
417
#### JSON Responses
418
419
```python
420
from quart import Quart, jsonify, request
421
import json
422
423
app = Quart(__name__)
424
425
@app.route('/api/users')
426
async def api_users():
427
users = await get_all_users()
428
return jsonify({
429
'users': users,
430
'count': len(users),
431
'status': 'success'
432
})
433
434
@app.route('/api/user/<int:user_id>')
435
async def api_user(user_id):
436
user = await get_user(user_id)
437
if not user:
438
return jsonify({'error': 'User not found'}), 404
439
440
return jsonify(user.to_dict())
441
442
@app.route('/api/search')
443
async def api_search():
444
query = request.args.get('q', '')
445
results = await search_database(query)
446
447
# Use jsonify with kwargs
448
return jsonify(
449
query=query,
450
results=results,
451
total=len(results),
452
timestamp=datetime.now().isoformat()
453
)
454
455
@app.route('/api/bulk-data')
456
async def bulk_data():
457
# Large dataset
458
data = await get_large_dataset()
459
460
# Custom JSON encoding
461
return jsonify(data), 200, {
462
'Content-Type': 'application/json; charset=utf-8',
463
'Cache-Control': 'max-age=300'
464
}
465
466
# Custom JSON encoder for complex objects
467
class CustomJSONEncoder(json.JSONEncoder):
468
def default(self, obj):
469
if isinstance(obj, datetime):
470
return obj.isoformat()
471
if isinstance(obj, Decimal):
472
return float(obj)
473
return super().default(obj)
474
475
app.json_encoder = CustomJSONEncoder
476
```
477
478
#### Advanced Features
479
480
```python
481
from quart import Quart, stream_with_context, make_push_promise, Response
482
import asyncio
483
484
app = Quart(__name__)
485
486
@app.route('/streaming-data')
487
async def streaming_data():
488
@stream_with_context
489
async def generate():
490
for i in range(100):
491
# Access request context during streaming
492
user_id = request.args.get('user_id')
493
data = await get_user_data(user_id, page=i)
494
495
yield f"data: {json.dumps(data)}\n\n"
496
await asyncio.sleep(0.1)
497
498
return Response(generate(), mimetype='text/plain')
499
500
@app.route('/with-push-promise')
501
async def with_push_promise():
502
# Push critical resources to client (HTTP/2)
503
await make_push_promise('/static/critical.css')
504
await make_push_promise('/static/critical.js')
505
506
return await render_template('page.html')
507
508
@app.route('/template-data/<template_name>')
509
async def template_data(template_name):
510
# Get template metadata without rendering
511
try:
512
title = get_template_attribute(f'{template_name}.html', 'title')
513
description = get_template_attribute(f'{template_name}.html', 'description')
514
515
return jsonify({
516
'template': template_name,
517
'title': title,
518
'description': description
519
})
520
except Exception as e:
521
return jsonify({'error': str(e)}), 404
522
523
@app.route('/server-sent-events')
524
async def server_sent_events():
525
@stream_with_context
526
async def event_stream():
527
yield "data: Connected to event stream\n\n"
528
529
while True:
530
# Real-time data streaming
531
event_data = await get_next_event()
532
yield f"data: {json.dumps(event_data)}\n\n"
533
534
await asyncio.sleep(1)
535
536
return Response(
537
event_stream(),
538
mimetype='text/event-stream',
539
headers={
540
'Cache-Control': 'no-cache',
541
'Connection': 'keep-alive'
542
}
543
)
544
```