A Python ASGI web framework with the same API as Flask
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
HTTP helpers, URL generation, file serving, flash messaging, JSON support, and advanced features for comprehensive web development workflows.
Core HTTP utilities for response creation, error handling, and request flow control.
def abort(code: int, *args, **kwargs) -> NoReturn:
"""
Raise HTTP exception with specified status code.
Args:
code: HTTP status code (400, 404, 500, etc.)
*args: Additional arguments for exception
**kwargs: Additional keyword arguments
Raises:
HTTPException: Always raises, never returns
"""
async def make_response(*args) -> Response:
"""
Create response object from various input types.
Args:
*args: Response data (string, dict, tuple, Response, etc.)
Returns:
Response object
"""
def redirect(location: str, code: int = 302, Response=None) -> Response:
"""
Create redirect response.
Args:
location: URL to redirect to
code: HTTP redirect status code (301, 302, 303, 307, 308)
Response: Custom response class (optional)
Returns:
Redirect response object
"""URL construction and endpoint resolution with support for external URLs and query parameters.
def url_for(endpoint: str, **values) -> str:
"""
Generate URL for endpoint with given parameters.
Args:
endpoint: Endpoint name (function name or 'blueprint.function_name')
**values: URL parameters and query string arguments
Returns:
Generated URL string
Example:
url_for('user_profile', username='john', page=2)
# Returns: '/user/john?page=2'
"""Async file serving with MIME type detection, caching headers, and download attachment support.
async def send_file(
filename_or_io,
mimetype: str | None = None,
as_attachment: bool = False,
download_name: str | None = None,
conditional: bool = True,
etag: bool | str = True,
last_modified: datetime | None = None,
max_age: int | None = None,
cache_timeout: int | None = None
) -> Response:
"""
Send file as response with proper headers.
Args:
filename_or_io: File path or file-like object
mimetype: MIME type (auto-detected if None)
as_attachment: Force download vs inline display
download_name: Filename for download
conditional: Enable conditional requests (304 responses)
etag: ETag value or True for auto-generation
last_modified: Last modified datetime
max_age: Cache max-age in seconds
cache_timeout: Deprecated, use max_age
Returns:
File response with appropriate headers
"""
async def send_from_directory(
directory: str,
filename: str,
**kwargs
) -> Response:
"""
Send file from directory with path safety checks.
Args:
directory: Base directory path
filename: Filename within directory (path traversal protected)
**kwargs: Additional arguments for send_file()
Returns:
File response
Raises:
NotFound: If file doesn't exist or path is invalid
"""Session-based flash messaging system for user notifications across requests.
async def flash(message: str, category: str = "message"):
"""
Add flash message to session for next request.
Args:
message: Message text to display
category: Message category ('message', 'error', 'warning', 'info')
"""
def get_flashed_messages(
with_categories: bool = False,
category_filter: tuple = ()
) -> list:
"""
Retrieve and clear flash messages from session.
Args:
with_categories: Return (category, message) tuples instead of just messages
category_filter: Only return messages from specified categories
Returns:
List of messages or (category, message) tuples
"""JSON response creation and data serialization with customizable encoding.
def jsonify(*args, **kwargs) -> Response:
"""
Create JSON response from data.
Args:
*args: Positional arguments (single value becomes response)
**kwargs: Keyword arguments (become JSON object)
Returns:
Response with JSON content and appropriate headers
Examples:
jsonify({'key': 'value'}) # JSON object
jsonify([1, 2, 3]) # JSON array
jsonify(key='value') # JSON object from kwargs
"""
def dumps(obj, **kwargs) -> str:
"""
Serialize object to JSON string using app's JSON encoder.
Args:
obj: Object to serialize
**kwargs: Additional arguments for json.dumps()
Returns:
JSON string
"""
def dump(obj, fp, **kwargs):
"""
Serialize object to JSON file using app's JSON encoder.
Args:
obj: Object to serialize
fp: File-like object to write to
**kwargs: Additional arguments for json.dump()
"""
def loads(s: str, **kwargs):
"""
Deserialize JSON string using app's JSON decoder.
Args:
s: JSON string to deserialize
**kwargs: Additional arguments for json.loads()
Returns:
Deserialized Python object
"""
def load(fp, **kwargs):
"""
Deserialize JSON file using app's JSON decoder.
Args:
fp: File-like object to read from
**kwargs: Additional arguments for json.load()
Returns:
Deserialized Python object
"""Specialized features for HTTP/2 push promises, streaming with context, and template integration.
async def make_push_promise(path: str):
"""
Create HTTP/2 server push promise.
Args:
path: Resource path to push to client
Note:
Only works with HTTP/2 compatible servers
"""
def stream_with_context(func: Callable):
"""
Decorator to stream response while preserving request context.
Args:
func: Generator function that yields response chunks
Returns:
Decorated function that maintains context during streaming
"""
def get_template_attribute(template_name: str, attribute: str):
"""
Get attribute from template without rendering.
Args:
template_name: Template file name
attribute: Attribute name to retrieve
Returns:
Template attribute value
"""from quart import Quart, abort, make_response, redirect, url_for
app = Quart(__name__)
@app.route('/user/<int:user_id>')
async def user_profile(user_id):
user = await get_user(user_id)
if not user:
abort(404) # Raises 404 Not Found
return f"User: {user.name}"
@app.route('/admin')
async def admin_area():
if not current_user.is_admin:
abort(403) # Raises 403 Forbidden
return "Admin Dashboard"
@app.route('/custom-response')
async def custom_response():
# Create custom response
response = await make_response("Custom content")
response.status_code = 201
response.headers['X-Custom'] = 'value'
return response
@app.route('/login', methods=['POST'])
async def login():
# Process login...
if login_successful:
return redirect(url_for('dashboard'))
else:
return redirect(url_for('login_form', error='invalid'))
@app.route('/old-url')
async def old_url():
# Permanent redirect
return redirect(url_for('new_url'), code=301)from quart import Quart, send_file, send_from_directory
app = Quart(__name__)
@app.route('/download/<filename>')
async def download_file(filename):
# Send file as download
return await send_from_directory(
'uploads',
filename,
as_attachment=True,
download_name=f"document_{filename}"
)
@app.route('/image/<image_id>')
async def serve_image(image_id):
# Serve image with caching
image_path = f"/images/{image_id}.jpg"
return await send_file(
image_path,
mimetype='image/jpeg',
max_age=3600 # Cache for 1 hour
)
@app.route('/report/pdf')
async def generate_report():
# Generate PDF and send
pdf_data = await generate_pdf_report()
return await send_file(
io.BytesIO(pdf_data),
mimetype='application/pdf',
as_attachment=True,
download_name=f'report_{datetime.now().date()}.pdf'
)
@app.route('/avatar/<username>')
async def user_avatar(username):
# Serve user avatar with fallback
avatar_path = f"/avatars/{username}.png"
if os.path.exists(avatar_path):
return await send_file(avatar_path, max_age=1800)
else:
# Send default avatar
return await send_file("/static/default_avatar.png")from quart import Quart, flash, get_flashed_messages, render_template, redirect, url_for
app = Quart(__name__)
app.secret_key = 'secret-key-for-sessions'
@app.route('/form', methods=['POST'])
async def process_form():
form_data = await request.form
try:
# Process form data
result = await process_user_data(form_data)
await flash(f'Successfully processed {result.count} items!', 'success')
await flash('Data has been saved to the database.', 'info')
except ValidationError as e:
await flash(f'Validation error: {e}', 'error')
except Exception as e:
await flash('An unexpected error occurred.', 'error')
return redirect(url_for('show_form'))
@app.route('/form')
async def show_form():
# Get flash messages for template
messages = get_flashed_messages(with_categories=True)
return await render_template('form.html', messages=messages)
@app.route('/admin/bulk-update', methods=['POST'])
async def bulk_update():
try:
updates = await perform_bulk_update()
for category, count in updates.items():
await flash(f'Updated {count} {category} records', 'success')
except Exception as e:
await flash('Bulk update failed', 'error')
await flash(str(e), 'error')
return redirect(url_for('admin_dashboard'))
# Template usage (form.html):
"""
{% for category, message in messages %}
<div class="alert alert-{{ category }}">{{ message }}</div>
{% endfor %}
"""from quart import Quart, jsonify, request
import json
app = Quart(__name__)
@app.route('/api/users')
async def api_users():
users = await get_all_users()
return jsonify({
'users': users,
'count': len(users),
'status': 'success'
})
@app.route('/api/user/<int:user_id>')
async def api_user(user_id):
user = await get_user(user_id)
if not user:
return jsonify({'error': 'User not found'}), 404
return jsonify(user.to_dict())
@app.route('/api/search')
async def api_search():
query = request.args.get('q', '')
results = await search_database(query)
# Use jsonify with kwargs
return jsonify(
query=query,
results=results,
total=len(results),
timestamp=datetime.now().isoformat()
)
@app.route('/api/bulk-data')
async def bulk_data():
# Large dataset
data = await get_large_dataset()
# Custom JSON encoding
return jsonify(data), 200, {
'Content-Type': 'application/json; charset=utf-8',
'Cache-Control': 'max-age=300'
}
# Custom JSON encoder for complex objects
class CustomJSONEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, datetime):
return obj.isoformat()
if isinstance(obj, Decimal):
return float(obj)
return super().default(obj)
app.json_encoder = CustomJSONEncoderfrom quart import Quart, stream_with_context, make_push_promise, Response
import asyncio
app = Quart(__name__)
@app.route('/streaming-data')
async def streaming_data():
@stream_with_context
async def generate():
for i in range(100):
# Access request context during streaming
user_id = request.args.get('user_id')
data = await get_user_data(user_id, page=i)
yield f"data: {json.dumps(data)}\n\n"
await asyncio.sleep(0.1)
return Response(generate(), mimetype='text/plain')
@app.route('/with-push-promise')
async def with_push_promise():
# Push critical resources to client (HTTP/2)
await make_push_promise('/static/critical.css')
await make_push_promise('/static/critical.js')
return await render_template('page.html')
@app.route('/template-data/<template_name>')
async def template_data(template_name):
# Get template metadata without rendering
try:
title = get_template_attribute(f'{template_name}.html', 'title')
description = get_template_attribute(f'{template_name}.html', 'description')
return jsonify({
'template': template_name,
'title': title,
'description': description
})
except Exception as e:
return jsonify({'error': str(e)}), 404
@app.route('/server-sent-events')
async def server_sent_events():
@stream_with_context
async def event_stream():
yield "data: Connected to event stream\n\n"
while True:
# Real-time data streaming
event_data = await get_next_event()
yield f"data: {json.dumps(event_data)}\n\n"
await asyncio.sleep(1)
return Response(
event_stream(),
mimetype='text/event-stream',
headers={
'Cache-Control': 'no-cache',
'Connection': 'keep-alive'
}
)Install with Tessl CLI
npx tessl i tessl/pypi-quart