Quart Async Flask — Deep Dive

Production Quart patterns: Flask migration strategies, WebSocket architecture, async extension compatibility, and Hypercorn deployment.

Application architecture

A production Quart application follows the same patterns as Flask, with async additions:

from quart import Quart
from quart_cors import cors

def create_app():
    app = Quart(__name__)
    app = cors(app, allow_origin="https://myapp.com")
    
    app.config.from_mapping(
        DATABASE_URL="postgresql://user:pass@localhost/db",
        REDIS_URL="redis://localhost",
        SECRET_KEY="your-secret-key",
    )
    
    # Register blueprints
    from blueprints.api import api_bp
    from blueprints.auth import auth_bp
    app.register_blueprint(api_bp, url_prefix="/api/v1")
    app.register_blueprint(auth_bp, url_prefix="/auth")
    
    # Lifecycle hooks
    @app.before_serving
    async def startup():
        app.db_pool = await asyncpg.create_pool(
            app.config["DATABASE_URL"]
        )
        app.redis = await aioredis.from_url(
            app.config["REDIS_URL"]
        )
    
    @app.after_serving
    async def shutdown():
        await app.db_pool.close()
        await app.redis.close()
    
    return app

Note before_serving/after_serving instead of Flask’s context-based approach — these are Quart-specific lifecycle hooks that run when the ASGI server starts and stops.

Flask migration step by step

A real-world migration from Flask to Quart involves several layers:

Phase 1: Direct port

# Before (Flask)
from flask import Flask, jsonify, request
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
db = SQLAlchemy(app)

@app.route("/users")
def list_users():
    users = User.query.all()
    return jsonify([u.to_dict() for u in users])

# After (Quart — minimal changes)
from quart import Quart, jsonify, request

app = Quart(__name__)

@app.route("/users")
async def list_users():
    # Still sync DB call — works but doesn't benefit from async
    users = User.query.all()
    return jsonify([u.to_dict() for u in users])

Quart can run synchronous handlers — it executes them in a thread pool. This lets you migrate incrementally without converting everything at once.

Phase 2: Async database layer

import asyncpg

@app.route("/users")
async def list_users():
    async with app.db_pool.acquire() as conn:
        rows = await conn.fetch("SELECT id, name, email FROM users")
        users = [dict(row) for row in rows]
    return jsonify(users)

This is where the real performance gain happens. Each await conn.fetch() releases the event loop to handle other requests while waiting for the database.

Phase 3: Async HTTP clients

import httpx

# Before: requests (blocking)
@app.route("/weather")
def get_weather():
    resp = requests.get("https://api.weather.com/current")
    return jsonify(resp.json())

# After: httpx (async)
@app.route("/weather")
async def get_weather():
    async with httpx.AsyncClient() as client:
        resp = await client.get("https://api.weather.com/current")
    return jsonify(resp.json())

WebSocket patterns

Quart’s WebSocket support enables real-time features without Flask-SocketIO:

from quart import Quart, websocket
import asyncio
import json

app = Quart(__name__)

# Connection manager
class ConnectionManager:
    def __init__(self):
        self.connections: dict[str, set] = {}
    
    async def connect(self, room: str, ws):
        if room not in self.connections:
            self.connections[room] = set()
        self.connections[room].add(ws)
    
    async def disconnect(self, room: str, ws):
        self.connections[room].discard(ws)
        if not self.connections[room]:
            del self.connections[room]
    
    async def broadcast(self, room: str, message: dict):
        if room in self.connections:
            dead = set()
            for ws in self.connections[room]:
                try:
                    await ws.send_json(message)
                except Exception:
                    dead.add(ws)
            self.connections[room] -= dead

manager = ConnectionManager()

@app.websocket("/ws/<room>")
async def ws_handler(room):
    await websocket.accept()
    ws = websocket._get_current_object()
    await manager.connect(room, ws)
    
    try:
        while True:
            data = await websocket.receive_json()
            data["room"] = room
            await manager.broadcast(room, data)
    except asyncio.CancelledError:
        pass
    finally:
        await manager.disconnect(room, ws)

For production, replace the in-memory ConnectionManager with a Redis pub/sub backend for multi-process support.

Server-Sent Events (SSE)

Quart handles SSE naturally through async generators:

from quart import Quart, Response
import asyncio
import json

@app.route("/events")
async def events():
    async def generate():
        queue = asyncio.Queue()
        app.event_subscribers.add(queue)
        try:
            while True:
                event = await queue.get()
                yield f"event: {event['type']}\n"
                yield f"data: {json.dumps(event['data'])}\n\n"
        except asyncio.CancelledError:
            pass
        finally:
            app.event_subscribers.discard(queue)
    
    headers = {
        "Content-Type": "text/event-stream",
        "Cache-Control": "no-cache",
        "Connection": "keep-alive",
    }
    return Response(generate(), headers=headers)

# Publishing events from any handler
@app.route("/api/notify", methods=["POST"])
async def notify():
    data = await request.get_json()
    event = {"type": "notification", "data": data}
    
    for queue in app.event_subscribers:
        await queue.put(event)
    
    return jsonify({"sent_to": len(app.event_subscribers)})

Extension compatibility

Many Flask extensions work with Quart directly or have async equivalents:

Flask Extension	Quart Status
Flask-Login	Works via quart-auth
Flask-CORS	Works via quart-cors
Flask-SQLAlchemy	Replaced by direct asyncpg/async SQLAlchemy
Flask-Caching	Works via quart-rate-limiter
Flask-WTF	Works with quart-wtf
Flask-Mail	Replaced by aiosmtplib
Flask-SocketIO	Not needed — native WebSocket support
Jinja2	Works directly (Quart uses Jinja2)

For extensions without async equivalents, Quart can run sync code in a thread pool:

from quart import Quart
import asyncio

@app.route("/legacy")
async def legacy_endpoint():
    # Run sync Flask extension code in thread pool
    result = await asyncio.to_thread(sync_extension.do_work)
    return jsonify(result)

Middleware patterns

Quart supports ASGI middleware alongside Flask-style before/after hooks:

# Flask-style hooks (familiar)
@app.before_request
async def before_request():
    request.start_time = time.perf_counter()

@app.after_request
async def after_request(response):
    elapsed = time.perf_counter() - request.start_time
    response.headers["X-Response-Time"] = f"{elapsed:.4f}"
    return response

# ASGI middleware (for advanced use)
from quart import Quart

class SecurityHeadersMiddleware:
    def __init__(self, app):
        self.app = app
    
    async def __call__(self, scope, receive, send):
        if scope["type"] == "http":
            async def send_with_headers(message):
                if message["type"] == "http.response.start":
                    headers = list(message.get("headers", []))
                    headers.append((b"x-content-type-options", b"nosniff"))
                    headers.append((b"x-frame-options", b"DENY"))
                    message["headers"] = headers
                await send(message)
            await self.app(scope, receive, send_with_headers)
        else:
            await self.app(scope, receive, send)

app = Quart(__name__)
app.asgi_app = SecurityHeadersMiddleware(app.asgi_app)

Testing

Quart provides an async test client:

import pytest
from app import create_app

@pytest.fixture
def app():
    app = create_app()
    app.config["TESTING"] = True
    return app

@pytest.mark.asyncio
async def test_list_users(app):
    async with app.test_client() as client:
        response = await client.get("/api/v1/users")
        assert response.status_code == 200
        data = await response.get_json()
        assert isinstance(data, list)

@pytest.mark.asyncio
async def test_websocket(app):
    async with app.test_client() as client:
        async with client.websocket("/ws/test-room") as ws:
            await ws.send_json({"text": "hello"})
            response = await ws.receive_json()
            assert "text" in response

Deployment with Hypercorn

Hypercorn is the recommended ASGI server for Quart:

# Basic
hypercorn app:app --bind 0.0.0.0:8000

# Production with workers
hypercorn app:app --workers 4 --bind 0.0.0.0:8000

# HTTP/2 with TLS
hypercorn app:app \
    --certfile cert.pem \
    --keyfile key.pem \
    --bind 0.0.0.0:443

# Unix socket (behind nginx)
hypercorn app:app --bind unix:/tmp/quart.sock

Hypercorn configuration file (hypercorn.toml):

bind = "0.0.0.0:8000"
workers = 4
accesslog = "-"
errorlog = "-"
graceful_timeout = 30
keep_alive_timeout = 120

Behind nginx with WebSocket proxying:

upstream quart {
    server unix:/tmp/quart.sock;
}

server {
    listen 443 ssl http2;
    
    location / {
        proxy_pass http://quart;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_set_header Host $host;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    }
}

Performance comparison

In benchmarks with similar configurations:

Simple JSON response: Quart is ~5-15% slower than raw Starlette/FastAPI due to Flask-compatibility overhead
Database-bound endpoints: Performance converges — the bottleneck is I/O, not framework overhead
WebSocket throughput: Comparable to other ASGI frameworks
Memory under concurrent load: Dramatically lower than Flask with threads (one event loop vs. hundreds of threads)

The performance tradeoff is clear: you pay a small overhead for Flask API compatibility, but gain massive concurrent connection handling improvements over Flask.

The one thing to remember: Quart’s value proposition is migration, not greenfield development — it’s the lowest-risk path to async Python for teams with existing Flask applications, providing native WebSocket and SSE support while keeping the Flask API that developers already know.

pythonweb-frameworksquartasyncflask