Quart: Python非同期Webフレームワークの詳細解説

Quartは、Pythonで書かれた比較的新しいWebマイクロフレームワークです。特に非同期処理 (async/await) をネイティブにサポートしている点が大きな特徴です。Quartは、非常に人気のあるWebフレームワークであるFlaskのAPIを再実装する形で開発されました。そのため、Flaskに慣れている開発者であれば、比較的スムーズにQuartへ移行したり、学習したりすることが可能です。

Quartを使うことで、以下のようなWebアプリケーションを構築できます:

• HTMLテンプレートのレンダリングと配信 (例: ブログサイト)
• RESTfulなJSON APIの開発
• WebSocketを利用したリアルタイム通信 (例: チャットサーバー)
• リクエストとレスポンスのストリーミング配信 (例: 動画配信)
• これらすべてを単一のアプリケーション内で実現

Quartは、非同期 (asyncio) ライブラリやコードだけでなく、従来の同期的なライブラリやコードも利用可能です。現代的なWebアプリケーション、特にI/Oバウンドな処理 (データベースアクセス、外部API呼び出しなど) が多いアプリケーションにおいて、非同期処理はパフォーマンスとスケーラビリティの向上に大きく貢献します。

QuartとFlaskは、APIの互換性が高いことからしばしば比較されます。両者は多くの共通点を持ちますが、根本的な違いも存在します。

どちらを選ぶべきか？

選択はプロジェクトの要件によります。

• Flaskが適しているケース:
  • 同期的な処理がメインのシンプルなWebアプリケーション
  • 豊富なドキュメントやコミュニティサポート、既存のFlask拡張を活用したい場合
  • 非同期処理の必要性が低い、またはパフォーマンス要件がそれほど厳しくない場合
  • 初心者で、日本語の情報が多い方が学習しやすい場合
• Quartが適しているケース:
  • 非同期処理 (async/await) を積極的に活用したい場合
  • 高いパフォーマンスとスケーラビリティが求められるI/Oバウンドなアプリケーション
  • WebSocketを使ったリアルタイム機能が必要な場合
  • Flaskの経験があり、非同期の世界へスムーズに移行したい場合
  • 主に非同期のコードベースを持っている、または構築する予定の場合

基本的に、多くのWebアプリケーション（データベースアクセスや外部API呼び出しを行うもの）は、非同期処理の恩恵を受けられるため、新規プロジェクトではQuartを検討する価値は高いと言えます。

インストール

Quartはpipを使って簡単にインストールできます。Python 3.7以降が必要です。

pip install quart

本番環境では、Quartアプリケーションを直接実行するのではなく、HypercornやUvicornなどのASGIサーバーを使用することが推奨されます。

pip install hypercorn # または pip install uvicorn

基本的なアプリケーション

非常にシンプルなQuartアプリケーションの例です。app.pyという名前で保存しましょう。

from quart import Quart, jsonify, render_template, websocket

# Quartアプリケーションインスタンスを作成
app = Quart(__name__)

# ルートURL ("/") へのGETリクエストを処理
@app.route("/")
async def index():
    # HTMLテンプレートをレンダリングして返す (非同期で)
    # templatesフォルダにindex.htmlを配置する必要がある
    # return await render_template("index.html")
    return "<h1>Hello, Quart!</h1>" # 簡単のためHTML文字列を直接返す

# "/api" へのGETリクエストを処理
@app.route("/api")
async def json_api():
    # JSONレスポンスを返す (非同期で)
    return jsonify({"message": "This is a JSON API response from Quart!"})

# "/ws" へのWebSocket接続を処理
@app.websocket("/ws")
async def ws_endpoint():
    try:
        while True:
            # クライアントからのメッセージを受信 (非同期で)
            data = await websocket.receive()
            print(f"Received from client: {data}")

            # クライアントにメッセージを送信 (非同期で)
            await websocket.send(f"Echo: {data}")

            # JSON形式でメッセージを送信することも可能
            # await websocket.send_json({"response": f"Received {data}"})
    except asyncio.CancelledError:
        # クライアント切断時の処理
        print("WebSocket client disconnected")
        raise

# スクリプトが直接実行された場合にサーバーを起動
if __name__ == "__main__":
    # 開発用サーバーを実行 (デバッグモード有効)
    # 本番環境ではASGIサーバー (Hypercorn, Uvicorn) を使用する
    app.run(debug=True, host="0.0.0.0", port=5000)

実行方法

開発中は、quart runコマンド（内部的にapp.run()を呼び出す）または直接スクリプトを実行します。

# quart run コマンドを使用 (推奨)
# QUART_APP環境変数にファイル名:アプリインスタンス名 を設定
export QUART_APP=app:app
quart run --host 0.0.0.0 --port 5000 --reload # --reloadでコード変更時に自動再読み込み

# または直接実行 (app.run() が呼ばれる)
python app.py

本番環境では、ASGIサーバーを使います。

# Hypercornを使用する場合
hypercorn app:app --bind 0.0.0.0:5000

# Uvicornを使用する場合
uvicorn app:app --host 0.0.0.0 --port 5000 --reload # 開発時は--reloadも便利

これで、ブラウザで http://localhost:5000/ にアクセスすると “Hello, Quart!” が、http://localhost:5000/api にアクセスするとJSONが表示されます。WebSocketのテストには、対応するクライアントツールが必要です。

Quartの基本的な構成要素はFlaskと似ていますが、非同期処理が前提となっている点が異なります。

ルーティング (Routing)

@app.route() デコレータを使って、特定のURLパスとHTTPメソッド（デフォルトはGET）に対応する関数（ビュー関数）を結びつけます。ビュー関数は async def で定義する必要があります。

from quart import Quart, request

app = Quart(__name__)

# GETリクエストを処理
@app.route('/user/<username>')
async def show_user_profile(username):
    # URLパスからパラメータを受け取る
    return f'User {username}'

# POSTリクエストを処理
@app.route('/login', methods=['POST'])
async def login():
    # リクエストデータにアクセス (非同期で)
    form_data = await request.form
    username = form_data.get('username')
    password = form_data.get('password')
    # ログイン処理...
    return f'Login attempt for {username}'

# GETとPOSTの両方を処理
@app.route('/contact', methods=['GET', 'POST'])
async def contact():
    if request.method == 'POST':
        # POST時の処理 (フォーム送信など)
        data = await request.form
        # ...
        return 'Message received!'
    else:
        # GET時の処理 (フォーム表示など)
        return '<form method="post"><input type="text" name="message"><button type="submit">Send</button></form>'

URL変数 (<username>) や、許可するHTTPメソッド (methods=['POST']) を指定できます。

リクエストとレスポンス (Request & Response Objects)

Flaskと同様に、リクエストコンテキスト内で request オブジェクトを通じて現在のリクエスト情報にアクセスできます。フォームデータ、JSONボディ、クエリパラメータ、ヘッダーなどにアクセスする際は、多くの場合 await が必要になります。

from quart import Quart, request, make_response, jsonify
import json

app = Quart(__name__)

@app.route('/process', methods=['POST'])
async def process_data():
    # クエリパラメータ (?key=value)
    param = request.args.get('param', 'default_value')

    # JSONボディの取得
    try:
        json_data = await request.get_json()
        if json_data is None:
            return await make_response(jsonify({"error": "Request must be JSON"}), 400)
        item = json_data.get('item')
    except json.JSONDecodeError:
         return await make_response(jsonify({"error": "Invalid JSON"}), 400)

    # フォームデータの取得
    # form_data = await request.form
    # name = form_data.get('name')

    # ヘッダーの取得
    user_agent = request.headers.get('User-Agent')

    # レスポンスの作成
    response_data = {"received_param": param, "received_item": item, "agent": user_agent}

    # make_responseで詳細なレスポンスを作成可能
    response = await make_response(jsonify(response_data), 200) # ステータスコード指定
    response.headers['X-Custom-Header'] = 'Quart-Value' # カスタムヘッダー追加
    return response

レスポンスは、文字列、jsonifyによるJSONオブジェクト、または make_response 関数で作成したレスポンスオブジェクトを返すことで生成します。テンプレートをレンダリングする場合は render_template を使います（後述）。

テンプレート (Templates)

Quartはデフォルトで Jinja2 テンプレートエンジンを使用します。render_template 関数を await してHTMLを生成します。テンプレートファイルは通常、アプリケーションルート直下の templates フォルダに置きます。

from quart import Quart, render_template

app = Quart(__name__)

@app.route('/hello/')
@app.route('/hello/<name>')
async def hello(name=None):
    # templates/hello.html をレンダリング
    return await render_template('hello.html', user_name=name)

templates/hello.html の例:

<!doctype html>
<title>Hello from Quart</title>
{% if user_name %}
  <h1>Hello, {{ user_name }}!</h1>
{% else %}
  <h1>Hello, World!</h1>
{% endif %}

render_template の第二引数以降で、テンプレート内で使用する変数を渡します。

ブループリント (Blueprints)

大規模なアプリケーションを開発する場合、関連するルートや機能をモジュール化するためにブループリントを使用します。これはFlaskのブループリントとほぼ同じ概念です。

例: 認証関連の機能をまとめる auth.py:

# auth.py
from quart import Blueprint, render_template, request, redirect, url_for

# Blueprintオブジェクトを作成
auth_bp = Blueprint('auth', __name__, url_prefix='/auth', template_folder='templates/auth')

@auth_bp.route('/login', methods=['GET', 'POST'])
async def login():
    if request.method == 'POST':
        # ログイン処理 ...
        return redirect(url_for('main.index')) # mainブループリントのindexへリダイレクト
    return await render_template('login.html') # templates/auth/login.html を探す

@auth_bp.route('/logout')
async def logout():
    # ログアウト処理 ...
    return redirect(url_for('main.index'))

アプリケーション本体 (app.py) でブループリントを登録します。

# app.py
from quart import Quart
from auth import auth_bp # auth.py からインポート
# 他のブループリントもインポート ... (e.g., from main import main_bp)

app = Quart(__name__)

# ブループリントを登録
app.register_blueprint(auth_bp)
# app.register_blueprint(main_bp)

# アプリケーション直下のルートも定義できる
@app.route('/')
async def index():
    return 'Welcome to the main page!'

if __name__ == "__main__":
    app.run()

url_prefix でブループリント内の全ルートに共通のプレフィックスを指定できます (例: /auth/login)。template_folder でブループリント専用のテンプレートフォルダを指定することも可能です。

コンテキスト (Contexts)

Flaskと同様に、Quartにもアプリケーションコンテキストとリクエストコンテキスト（およびWebSocketコンテキスト）があります。これらは、リクエスト処理中に特定の変数（request, session, websocket など）にアクセスできるようにするための仕組みです。Quartではこれらのコンテキストも非同期に対応しています。

セッション (Sessions)

QuartはFlaskと同様に、安全なクッキーベースのセッションを提供します。session オブジェクト（辞書ライク）にデータを格納すると、そのデータは署名付きクッキーとしてクライアントに保存され、次回のリクエストで復元されます。セッションを利用するには、app.secret_key を設定する必要があります。

from quart import Quart, session, redirect, url_for, request, render_template
import os

app = Quart(__name__)
# 安全なキーを設定 (環境変数などから取得するのが望ましい)
app.secret_key = os.environ.get('SECRET_KEY', 'a_very_secret_key_for_dev')

@app.route('/set_user/<username>')
async def set_user(username):
    session['username'] = username # セッションにユーザー名を保存
    return f"Username '{username}' set in session."

@app.route('/get_user')
async def get_user():
    username = session.get('username') # セッションからユーザー名を取得
    if username:
        return f"Logged in as: {username}"
    else:
        return "Not logged in."

@app.route('/clear_user')
async def clear_user():
    session.pop('username', None) # セッションからユーザー名を削除
    return "Username removed from session."

セッションデータは署名されますが、暗号化はされないため、機密情報を直接保存するのは避けるべきです。

Quartの真価は非同期処理にあります。

async / await

Quartのビュー関数、WebSocketハンドラ、各種フック (before_request など) は async def で定義します。これにより、関数内で await を使って非同期I/O操作 (データベースアクセス、HTTPリクエスト、ファイルの読み書きなど) を待機できます。await 中はイベントループが他のタスクを実行できるため、ブロックすることなく効率的にリクエストを処理できます。

import asyncio
import httpx # 非同期HTTPクライアントライブラリ
from quart import Quart

app = Quart(__name__)

# 非同期HTTPクライアントのインスタンス
client = httpx.AsyncClient()

@app.route('/fetch_data')
async def fetch_data():
    try:
        # 外部APIに非同期でリクエスト (I/Oバウンドな処理)
        response = await client.get('https://httpbin.org/delay/1') # 1秒待つAPI
        response.raise_for_status() # エラーチェック
        data = response.json()

        # 時間のかかる可能性のある計算や処理 (CPUバウンドなら別スレッド/プロセス推奨)
        # ここでは簡単な例として asyncio.sleep を使用
        await asyncio.sleep(0.5) # 0.5秒待機 (他の処理に制御を譲る)
        processed_result = f"Data fetched: {data.get('origin')}, processed after sleep."

        return processed_result
    except httpx.RequestError as e:
        return f"HTTP Request failed: {e}", 500
    finally:
        # 必要ならクライアントを閉じる (アプリケーション全体で共有する場合は不要)
        # await client.aclose()
        pass

非同期ライブラリ (例: httpx, asyncpg, aiofiles) と組み合わせることで、Quartの非同期性能を最大限に引き出すことができます。

WebSocketサポート

QuartはWebSocketを標準でサポートしており、@app.websocket() デコレータでハンドラを定義します。ハンドラ内では websocket グローバルオブジェクトを通じて、メッセージの送受信 (send, receive, send_json, receive_json) や接続の受け入れ (accept) を非同期で行えます。

from quart import Quart, websocket, render_template
import asyncio

app = Quart(__name__)

connected_clients = set()

@app.route('/chat')
async def chat_page():
    # チャットページのHTMLを返す
    return await render_template('chat.html') # templates/chat.html が必要

@app.websocket('/ws_chat')
async def chat_ws():
    global connected_clients
    # 現在のWebSocket接続をセットに追加
    current_ws = websocket._get_current_object()
    connected_clients.add(current_ws)
    print(f"Client connected: {current_ws}. Total clients: {len(connected_clients)}")

    try:
        # 接続が続く限りメッセージを受信
        while True:
            message = await websocket.receive()
            print(f"Message received: {message}")

            # 接続されている全クライアントにメッセージをブロードキャスト
            broadcast_tasks = []
            for client_ws in connected_clients:
                # 自分自身には送らない場合
                # if client_ws != current_ws:
                broadcast_tasks.append(client_ws.send(f"User says: {message}"))

            # 全ての送信タスクを並行して実行
            await asyncio.gather(*broadcast_tasks)

    except asyncio.CancelledError:
        print("Client disconnected by CancelledError")
        # エラー処理 (接続がキャンセルされた場合)
    except Exception as e:
        print(f"An error occurred: {e}")
    finally:
        # クライアントが切断されたらセットから削除
        connected_clients.remove(current_ws)
        print(f"Client disconnected: {current_ws}. Remaining clients: {len(connected_clients)}")

上記の例では、接続された全クライアントにメッセージをブロードキャストする簡単なチャット機能を示しています。QuartはWebSocket接続の受け入れを制御する機能も提供しており、例えば認証が必要なWebSocketエンドポイントなども実装できます。

レスポンスストリーミング

大きなファイルやリアルタイムで生成されるデータを効率的に送信するために、レスポンスストリーミングが利用できます。非同期ジェネレータを使ってレスポンスボディをチャンクごとに生成し、送信します。

from quart import Quart, Response
import asyncio
import datetime

app = Quart(__name__)

async def generate_data():
    """非同期ジェネレータ: 1秒ごとに時刻を生成"""
    for i in range(10):
        yield f"data: {datetime.datetime.now().isoformat()}\n\n" # Server-Sent Events形式
        await asyncio.sleep(1)

@app.route('/stream')
async def stream_response():
    # ジェネレータをレスポンスボディとして指定
    response = await app.make_response(generate_data())
    response.mimetype = 'text/event-stream' # Server-Sent Events用MIMEタイプ
    response.timeout = None # タイムアウトを無効化
    return response

この例では、Server-Sent Events (SSE) 形式で1秒ごとにデータをクライアントにプッシュしています。

テスト (Testing)

Quartはテストクライアントを提供しており、アプリケーションを実際に起動せずにテストを実行できます。pytest などのテストフレームワークと組み合わせて使用するのが一般的です。非同期コードのテストには pytest-asyncio プラグインが役立ちます。

# tests/test_app.py
import pytest
from app import app # テスト対象のQuartアプリをインポート

@pytest.fixture(name="client")
def _client():
    # テストクライアントをセットアップ
    return app.test_client()

@pytest.mark.asyncio # pytest-asyncioが必要
async def test_index(client):
    response = await client.get('/')
    assert response.status_code == 200
    data = await response.get_data(as_text=True)
    assert "Hello, Quart!" in data

@pytest.mark.asyncio
async def test_json_api(client):
    response = await client.get('/api')
    assert response.status_code == 200
    assert response.mimetype == 'application/json'
    json_data = await response.get_json()
    assert json_data['message'] == "This is a JSON API response from Quart!"

@pytest.mark.asyncio
async def test_websocket(client):
    test_message = "ping"
    async with client.websocket('/ws') as ws:
        await ws.send(test_message)
        received_message = await ws.receive()
        assert received_message == f"Echo: {test_message}"

テストクライアントを使うことで、GET, POSTリクエストやWebSocket通信をシミュレートし、レスポンスや動作を検証できます。

拡張機能 (Extensions)

Flaskと同様に、Quartにも拡張機能のエコシステムがあります。データベース連携、認証、CORS対応など、特定の機能を追加するためのライブラリが存在します。一部のFlask拡張はQuartでも動作することがあります。

• Quart-Cors: CORS (Cross-Origin Resource Sharing) ヘッダーを管理します。
• Quart-SQLAlchemy: 非同期対応のSQLAlchemyインテグレーション (または他の非同期ORM/ライブラリ)。
• Quart-Auth: 認証機能を提供します。
• その他、多くのコミュニティ製拡張があります。

拡張機能を使うことで、共通の機能を再実装する手間を省き、開発を効率化できます。

デプロイメント (Deployment)

開発中は quart run や app.run() が便利ですが、本番環境では ASGI サーバーを使用する必要があります。代表的な ASGI サーバーには以下のようなものがあります。

• Hypercorn: QuartプロジェクトによってメンテナンスされているASGIサーバー。Trioやuvloopなどのイベントループもサポート。
• Uvicorn: FastAPIなどで広く使われている高速なASGIサーバー。uvloopをデフォルトで使用（Windows以外）。
• Daphne: Django Channelsプロジェクトの一部として開発されたASGIサーバー。

これらのサーバーは、Gunicornなどのプロセス管理ツールと組み合わせて、複数のワーカープロセスを起動し、スケーラビリティと耐障害性を高めることが一般的です。

# Gunicorn + Uvicornワーカーでデプロイする例
pip install gunicorn uvicorn

# 4つのワーカープロセスを起動
gunicorn -w 4 -k uvicorn.workers.UvicornWorker app:app -b 0.0.0.0:8000

Nginxなどのリバースプロキシを前段に配置し、静적ファイルの配信、HTTPS終端、負荷分散などを行う構成が一般的です。

設定 (Configuration)

アプリケーションの設定 (データベース接続情報、秘密鍵など) は、ファイルや環境変数から読み込むことができます。Flaskと同様の方法が利用可能です。

from quart import Quart
import os

app = Quart(__name__)

# デフォルト設定
app.config['DEBUG'] = False
app.config['SECRET_KEY'] = 'default_secret'

# 設定ファイルから読み込む (例: config.py)
# config.py に DEBUG = True などと記述
# app.config.from_pyfile('config.py', silent=True)

# 環境変数から読み込む
app.config['DATABASE_URL'] = os.environ.get('DATABASE_URL')
if os.environ.get('QUART_ENV') == 'development':
    app.config['DEBUG'] = True
    app.secret_key = 'dev_secret_key' # 環境変数などで上書き
else:
    app.secret_key = os.environ.get('SECRET_KEY') # 本番用キー

# 設定値へのアクセス
# db_url = app.config['DATABASE_URL']

環境変数 QUART_ENV (development または production) や QUART_DEBUG (1 または 0) も参照されます。

Quartは、Flaskの使いやすさと親しみやすさを維持しながら、Pythonの強力な非同期機能 (async/await) を活用できるようにしたモダンなWebフレームワークです。 ASGI標準に準拠し、WebSocketやHTTP/2をネイティブにサポートしているため、リアルタイム性が求められるアプリケーションや、高いパフォーマンス、スケーラビリティが必要なI/Oバウンドなアプリケーションに適しています。

Flaskからの移行も比較的容易であり、Pythonで非同期Webアプリケーションを開発する際の有力な選択肢の一つです。プロジェクトの要件に応じて、Flaskと比較検討し、Quartがもたらす非同期処理のメリットを活かせるかどうかを判断することが重要です。

もしあなたがFlaskに慣れていて、非同期プログラミングの世界に足を踏み入れたいと考えているなら、Quartは非常に良い出発点となるでしょう！ Quart公式ドキュメント もぜひ参照してみてください。