CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-quart

A Python ASGI web framework with the same API as Flask

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview
Eval results
Files

helpers.mddocs/

Helper Functions and Utilities

HTTP helpers, URL generation, file serving, flash messaging, JSON support, and advanced features for comprehensive web development workflows.

Capabilities

HTTP Response Helpers

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 Generation

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'
    """

File Serving

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
    """

Flash Messaging

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 Support

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
    """

Advanced Features

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
    """

Usage Examples

HTTP Response Helpers

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)

File Serving

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")

Flash Messaging

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 %}
"""

JSON Responses

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 = CustomJSONEncoder

Advanced Features

from 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

docs

context.md

core-application.md

helpers.md

index.md

request-response.md

signals.md

templates.md

testing.md

websocket.md

tile.json