🐍 Python ASGI Webフレームワーク「Responder」の使い方ガイド 🚀

セキュリティツール
ASGIPythonResponderWebフレームワーク使い方

はじめに：Responderとは？ 🤔

Responderは、Pythonコミュニティで有名な開発者であるKenneth Reitz氏によって開発されたWebサービスフレームワークです。彼は「Requests」や「Pipenv」といった人気ライブラリの作者としても知られています。Responderは、特に「人間にとって使いやすい」ことを目指して設計されました。

このフレームワークの基本的な考え方は、当時人気だったFlaskとFalconという二つのフレームワークの良いところを取り入れ、さらに新しいアイデアを加えて統合することでした。また、Requestsライブラリで培われたAPIの設計思想をWebフレームワークにも持ち込むことを目指していました。

ResponderはASGI (Asynchronous Server Gateway Interface) に準拠しています。ASGIは、WSGI (Web Server Gateway Interface) の後継として登場したインターフェースで、非同期処理を前提として設計されています。これにより、WebSocketやHTTP/2といったモダンなプロトコルを効率的に扱うことが可能になります。Responderは内部的にStarletteという高性能なASGIツールキットを利用しており、非同期処理を容易に実装できる点が大きな特徴です。

注意点： ResponderはKenneth Reitz氏自身によって「実験的なプロジェクト」であり、「学術的な演習」と位置付けられています。現在では、FastAPIのような、より成熟し、活発にメンテナンスされている代替フレームワークの使用が推奨されています。Responderは開発が活発ではなく、将来的なサポートに懸念があります。しかし、その設計思想やコードは学ぶべき点が多く、ASGIフレームワークの初期の試みとして興味深い存在です。

Responderの主な特徴 ✨

  • シンプルなAPI: import responder だけで主要な機能を利用できます。
  • クラスベースビュー: 継承なしでクラスを用いたビューを作成できます。
  • ASGI準拠: Python Webサービスの未来であるASGIに完全対応しています。
  • WebSocketサポート: リアルタイム双方向通信を簡単に実装できます。
  • f-string形式のルーティング: Python 3.6以降でお馴染みのf-string構文で直感的にルートを定義できます。
  • 変更可能なレスポンスオブジェクト: 各ビュー関数にrespオブジェクトが渡され、これを変更することでレスポンスを構築します。関数から値をreturnする必要はありません。
  • バックグラウンドタスク: ThreadPoolExecutorを利用して、レスポンス返却後に非同期でタスクを実行できます。
  • GraphQLサポート: Grapheneライブラリと連携し、GraphQL APIを構築できます（GraphiQLインターフェース付き）。
  • OpenAPIスキーマ自動生成: API定義から自動でOpenAPI（Swagger）仕様を生成し、インタラクティブなドキュメントを提供します。
  • 静的ファイル配信: WhiteNoiseが組み込まれており、プロダクション環境での静的ファイル配信に対応しています。
  • Jinja2テンプレート: 追加のインポートなしでJinja2テンプレートエンジンを利用できます。
  • 組み込みWebサーバー: uvloopをベースにした高パフォーマンスなASGIサーバー (Uvicorn) が組み込まれており、自動的にgzip圧縮も行います。
  • WSGI/ASGIアプリのマウント: FlaskやStarletteなど、他のWSGI/ASGIアプリケーションをサブルートにマウントできます。

インストール 💻

Responderを使用するには、Python 3.6以上が必要です（ドキュメントによっては3.7+と記載されている場合もありますが、初期は3.6+でした）。インストールはpipを使って簡単に行えます。

pip install responder

GraphQLやOpenAPIサポートなど、すべての拡張機能を含めてインストールする場合は、[full]オプションを使用します。

pip install 'responder[full]'

特定の機能のみを追加することも可能です。

# GraphQLサポートのみ
pip install 'responder[graphql]'

# OpenAPIサポートのみ
pip install 'responder[openapi]'

開発環境の管理には、Kenneth Reitz氏が開発したPipenvを使うと便利です。

# Pipenvをインストール（未導入の場合）
brew install pipenv  # macOSの場合 (他のOSでは適宜変更)

# プロジェクトディレクトリを作成し移動
mkdir myresponderapp
cd myresponderapp

# Python 3.7 (またはそれ以降) を指定して仮想環境を作成し、responderをインストール
pipenv install responder --python 3.7

基本的な使い方：Hello World! 👋

Responderで最も簡単なアプリケーションを作成してみましょう。以下のコードをapp.pyのような名前で保存します。

import responder

# Webサービス(API)オブジェクトを作成
api = responder.API()

# ルート "/" に対するビュー関数を定義
@api.route("/")
def hello_world(req, resp):
    resp.text = "hello, world!"

# サーバーを起動 (スクリプトとして直接実行された場合)
if __name__ == "__main__":
    api.run()

このコードは、まずresponder.API()でAPIインスタンスを作成します。次に、@api.route("/")デコレータを使って、ルートパス (/) へのGETリクエストを処理する関数hello_worldを定義します。

ビュー関数は、req (リクエストオブジェクト) と resp (レスポンスオブジェクト) の2つの引数を受け取ります。Responderでは、関数内でrespオブジェクトの属性（ここではresp.text）を変更することでレスポンス内容を決定します。関数から何かをreturnする必要はありません。

最後に、if __name__ == "__main__":ブロック内でapi.run()を呼び出すことで、開発用のWebサーバーを起動します。

ターミナルでこのファイルを実行します。

python app.py

デフォルトでは、http://127.0.0.1:5042でサーバーが起動します。Webブラウザでこのアドレスにアクセスすると、”hello, world!”というテキストが表示されるはずです。ポート番号を変更したい場合は、api.run(port=8000)のように指定できます。

ルーティング 🧭

Responderのルーティングは非常に直感的です。@api.route()デコレータを使用し、パスとビュー関数を結びつけます。

パスパラメータ

URLの一部を動的に変化させたい場合は、f-stringのような構文を使用します。

import responder

api = responder.API()

@api.route("/hello/{who}")
def hello_to(req, resp, *, who):
    resp.text = f"hello, {who}!"

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

この例では、/hello/リポジトリのようなURLにアクセスすると、whoパラメータにリポジトリという文字列が渡され、レスポンスは “hello, リポジトリ!” となります。

パラメータには型ヒントを指定することも可能です。サポートされているのはstr (デフォルト)、int、floatです。

@api.route("/add/{a:int}/{b:int}")
async def add(req, resp, *, a, b):
    result = a + b
    resp.text = f"{a} + {b} = {result}"

/add/5/3にアクセスすると、aには整数5、bには整数3が渡され、レスポンスは “5 + 3 = 8” となります。もし/add/hello/worldのように型に合わない値が渡された場合、Responderは自動的に404 Not Foundエラーを返します。

HTTPメソッド

デフォルトでは、@api.route()はGETリクエストのみを受け付けます。他のHTTPメソッド（POST, PUT, DELETEなど）を処理するには、methods引数を指定します。

@api.route("/items", methods=["POST"])
async def create_item(req, resp):
    # POSTリクエストの処理 (データ受信については後述)
    data = await req.media()
    print(f"Received data: {data}")
    resp.media = {"status": "item created", "data": data}
    resp.status_code = 201 # Created

@api.route("/items/{item_id}", methods=["PUT", "DELETE"])
async def update_or_delete_item(req, resp, *, item_id):
    if req.method == "put":
        # PUTリクエストの処理
        data = await req.media()
        resp.media = {"status": f"item {item_id} updated", "data": data}
    elif req.method == "delete":
        # DELETEリクエストの処理
        resp.media = {"status": f"item {item_id} deleted"}
        resp.status_code = 204 # No Content

req.method属性で、どのHTTPメソッドでリクエストされたかを確認できます。

クラスベースビュー

関連するHTTPメソッドの処理を一つのクラスにまとめることもできます。クラス内のメソッド名がHTTPメソッドに対応します（例: on_get, on_post）。

import responder

api = responder.API()

@api.route("/users/{user_id}")
class UserResource:
    def on_get(self, req, resp, *, user_id):
        resp.media = {"user_id": user_id, "name": f"User {user_id}"}

    async def on_put(self, req, resp, *, user_id):
        data = await req.media()
        resp.media = {"status": f"user {user_id} updated", "data": data}

    def on_delete(self, req, resp, *, user_id):
        resp.status_code = 204 # No Content

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

すべてのメソッド (GET, POST, PUT, DELETEなど) で共通の処理を行いたい場合は、on_requestメソッドを定義します。これはFalconフレームワークの考え方に似ています。

リクエストとレスポンス 📨📤

リクエスト (req)

ビュー関数に渡されるreqオブジェクトを通して、受信したリクエストに関する情報にアクセスできます。

  • req.method: HTTPメソッド (例: “GET”, “POST”)
  • req.url: 完全なURL
  • req.headers: リクエストヘッダー (Requestsライブラリと同様のケースインセンシティブな辞書)
  • req.query_params: クエリパラメータ (例: /search?q=hello の場合 {'q': 'hello'})
  • req.cookies: クッキー

リクエストボディのデータ（フォームデータ、JSONなど）を受け取る場合は、ビュー関数をasync defで定義し、await req.media()を使用する必要があります。ResponderはContent-Typeヘッダーに基づいて自動的にデータをパースします。

@api.route("/submit", methods=["POST"])
async def handle_submission(req, resp):
    # Content-Typeに応じてフォームデータ、JSON、YAMLを自動解析
    data = await req.media()

    # Content-Typeが 'application/x-www-form-urlencoded' または 'multipart/form-data' の場合
    # form_data = await req.form()

    # Content-Typeが 'application/json' の場合
    # json_data = await req.media() or await req.json

    # 生のバイト列としてボディを取得
    # raw_body = await req.content

    # テキストとしてボディを取得
    # text_body = await req.text

    resp.media = {"received": data}

req.media()は、一般的なContent-Type (JSON, YAML, Form) を自動判別してPythonオブジェクト（通常は辞書）に変換します。特定の形式を期待する場合は、req.json, req.yaml, req.formを直接使うこともできます。

レスポンス (resp)

respオブジェクトの属性を設定することで、クライアントに返すレスポンスを構築します。

  • resp.text = "...": テキスト (text/plain) を返します。
  • resp.html = "<html>...</html>": HTML (text/html) を返します。
  • resp.media = {...} or [...]: Pythonの辞書やリストをJSON (application/json) またはYAML (application/x-yaml、クライアントがAccept: application/x-yamlヘッダーを送った場合) に変換して返します。
  • resp.content = b"...": バイト列 (application/octet-streamなど、Content-Typeは別途設定推奨) を返します。
  • resp.status_code = ...: HTTPステータスコードを設定します (例: 200, 201, 404)。デフォルトは200 OKです。responder.status_codesモジュールに定数が用意されています (例: api.status_codes.HTTP_201_CREATED)。
  • resp.headers["Header-Name"] = "Value": レスポンスヘッダーを設定します。
  • resp.cookies["cookie_name"] = "value": クッキーを設定します。

例：JSONレスポンスを返す

@api.route("/info")
def get_info(req, resp):
    info_data = {
        "framework": "Responder",
        "version": "2.0.7", # 例
        "status": "Experimental"
    }
    resp.media = info_data
    resp.headers["X-Powered-By"] = "Responder Framework"

例：リダイレクトを行う

@api.route("/old-page")
def redirect_to_new(req, resp):
    resp.status_code = api.status_codes.HTTP_301_MOVED_PERMANENTLY
    resp.headers["Location"] = "/new-page"

テンプレートエンジン (Jinja2) 🎨

ResponderはJinja2テンプレートエンジンを組み込みでサポートしており、HTMLを動的に生成するのに便利です。追加のインポートは基本的に不要です。

まず、プロジェクトルートにtemplatesというディレクトリを作成し、その中にHTMLテンプレートファイルを配置します（例: templates/index.html）。

templates/index.html:

<!DOCTYPE html>
<html>
<head>
    <title>Hello {{ name }}!</title>
    <link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/bulma@1.0.1/css/bulma.min.css">
</head>
<body>
    <section class="section">
        <div class="container">
            <h1 class="title">Hello, {{ name }}!</h1>
            <p class="subtitle">This page is rendered by Responder with Jinja2.</p>
        </div>
    </section>
</body>
</html>

Pythonコード側では、resp.htmlにapi.template()メソッドの呼び出し結果を代入します。

import responder

api = responder.API()

@api.route("/greet/{name}")
def greet_html(req, resp, *, name):
    # templates/index.html をレンダリングし、name変数を渡す
    resp.html = api.template("index.html", name=name)

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

/greet/あなたの名前 にアクセスすると、templates/index.htmlがレンダリングされ、{{ name }}の部分がURLで指定した名前に置き換えられたHTMLが表示されます。

非同期でテンプレートをレンダリングする必要がある場合（テンプレート内で非同期関数を呼び出すなど）は、Templatesクラスを明示的にインスタンス化し、render_asyncを使用します。

from responder.templates import Templates

# 非同期レンダリングを有効にする
templates = Templates(directory="templates", enable_async=True)

@api.route("/async-greet/{name}")
async def async_greet_html(req, resp, *, name):
    # 非同期でレンダリング
    resp.html = await templates.render_async("index.html", name=name)

静的ファイル 🖼️📄

CSS、JavaScript、画像などの静的ファイルを配信するには、プロジェクトルートにstaticというディレクトリを作成し、その中にファイルを配置します。ResponderはWhiteNoiseライブラリを内部で使用しており、特別な設定なしで/static/パス以下でこれらのファイルを提供します。

例: static/css/style.css を作成した場合

/* static/css/style.css */
body {
    font-family: sans-serif;
    background-color: #f0f0f0;
}

h1 {
    color: #333;
}

HTMLテンプレートからは、/static/css/style.cssというパスで参照できます。

<head>
    <link rel="stylesheet" href="/static/css/style.css">
</head>

静的ファイルを提供するディレクトリ名やURLパスを変更したい場合は、responder.APIの初期化時に引数を指定します。

api = responder.API(static_dir="assets", static_route="/files")
# この場合、'assets' ディレクトリ内のファイルが '/files/' パスで提供される

バックグラウンドタスク ⏳

リクエストへのレスポンスを返した後に、時間のかかる処理（メール送信、データ処理など）を実行したい場合があります。Responderでは、@api.background.taskデコレータを使ってバックグラウンドタスクを簡単に定義できます。

バックグラウンドタスクを実行する関数を定義し、ビュー関数内でその関数を呼び出します。バックグラウンドタスクを実行するビュー関数はasync defである必要があります。

import responder
import time

api = responder.API()

@api.background.task
def process_data(data):
    """
    時間のかかる処理をシミュレート（例: 3秒待機）
    """
    print(f"Processing data in background: {data}")
    time.sleep(3)
    print("Background task finished.")

@api.route("/upload", methods=["POST"])
async def upload_file(req, resp):
    # リクエストからデータを取得
    data = await req.media()

    # バックグラウンドタスクとして process_data を呼び出す
    # この呼び出しはすぐに完了し、レスポンス処理に進む
    process_data(data)

    # クライアントにはすぐにレスポンスを返す
    resp.media = {"message": "File upload accepted, processing in background."}
    resp.status_code = 202 # Accepted

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

この例では、/uploadエンドポイントにPOSTリクエストを送ると、データを受け取った後、process_data関数がバックグラウンドで実行されます。クライアントにはすぐにHTTP 202 Acceptedレスポンスが返され、待たされることはありません。サーバーのコンソールには、バックグラウンドタスクの開始と終了のメッセージが（3秒の間隔をあけて）表示されます。

これはStarletteのBackgroundTasks機能を利用しています。FastAPIなど他のStarletteベースのフレームワークでも同様の機能が提供されています。

WebSocket ↔️

ResponderはWebSocketをサポートしており、リアルタイムの双方向通信アプリケーションを構築できます。WebSocketのエンドポイントは@api.routeデコレータで定義しますが、ビュー関数ではなく、WebSocket接続を処理するクラス（またはASGIアプリケーション）を指定します。

内部的にはStarletteのWebSocketサポートを利用します。以下は簡単なPing-Pongの例です。

import responder
from starlette.websockets import WebSocket, WebSocketDisconnect

api = responder.API(debug=True) # デバッグモードで詳細ログ表示

@api.route("/ws", websocket=True)
async def websocket_endpoint(ws: WebSocket):
    await ws.accept()
    print("WebSocket connected")
    try:
        while True:
            # クライアントからメッセージを受信
            data = await ws.receive_text()
            print(f"Received message: {data}")

            # メッセージに応じて応答
            if data == "ping":
                await ws.send_text("pong")
            else:
                await ws.send_text(f"Message received: {data}")
    except WebSocketDisconnect:
        print("WebSocket disconnected")
    except Exception as e:
        print(f"WebSocket error: {e}")
        await ws.close(code=1011) # Internal Server Error

if __name__ == "__main__":
    # api.run() は開発用サーバーであり、WebSocketの安定性に限界がある場合がある
    # プロダクションではUvicornを直接使うことを推奨
    # 例: uvicorn app:api --reload
    api.run()

この例では:

  1. @api.route("/ws", websocket=True)でWebSocketエンドポイントを定義します。
  2. 関数websocket_endpointは引数としてWebSocketオブジェクト (Starletteのもの) を受け取ります。
  3. await ws.accept()でクライアントからの接続要求を受け入れます。
  4. while Trueループ内でawait ws.receive_text()を呼び出し、クライアントからのテキストメッセージを待ち受けます。
  5. 受信したメッセージが “ping” なら “pong” を、それ以外なら受信したメッセージを含む応答をawait ws.send_text()でクライアントに送信します。
  6. クライアントが切断するとWebSocketDisconnect例外が発生し、ループを抜けます。
  7. 予期せぬエラーが発生した場合もエラーログを出力し、接続を閉じます。

このエンドポイントに接続するクライアント（例: JavaScript）を作成することで、リアルタイム通信が可能になります。

注意: api.run()で起動する開発サーバーは、WebSocketの長時間接続や多数の同時接続には向いていない場合があります。プロダクション環境では、uvicorn app:api --reloadのようにUvicorn ASGIサーバーを直接使用することが推奨されます。

より複雑なケースでは、Starletteのドキュメントにあるように、WebSocketの接続、受信、送信、切断を処理するクラスを定義し、api.add_route("/ws", YourWebSocketClass)のようにルートに追加する方法もあります。

テスト 🧪

Responderは、テスト用にRequestsライブラリに基づいたテストクライアントを提供します。これにより、実際のHTTPリクエストを送るのと同じような感覚でAPIをテストできます。

テストにはpytestなどのテストフレームワークを使用するのが一般的です。

例: test_app.py

import pytest
import app # app.py で api = responder.API() が定義されていると仮定

# pytestフィクスチャでAPIクライアントを準備
@pytest.fixture
def api_client():
    return app.api.requests # responder APIオブジェクトからテストクライアントを取得

def test_hello_world(api_client):
    """ "/" ルートが "hello, world!" を返すことをテスト """
    response = api_client.get("/")
    assert response.status_code == 200
    assert response.text == "hello, world!"

def test_hello_to(api_client):
    """ "/hello/{who}" ルートが正しく動作することをテスト """
    response = api_client.get("/hello/tester")
    assert response.status_code == 200
    assert response.text == "hello, tester!"

def test_post_data(api_client):
    """ POSTリクエストとJSONレスポンスをテスト """
    payload = {"key": "value", "number": 123}
    response = api_client.post("/submit", json=payload) # /submit がJSONを受け付けると仮定
    assert response.status_code == 200
    assert response.json()["received"] == payload

# 非同期エンドポイントのテストも同様に行える
# def test_async_endpoint(api_client):
#     response = api_client.get("/async-route")
#     assert response.status_code == 200
#     # ... アサーション ...

テストを実行するには、pytestコマンドを使用します。

# 開発依存ライブラリとしてpytestをインストール
pipenv install pytest --dev

# テストを実行
pipenv run pytest

api.requestsは、実際のRequestsライブラリと同様のインターフェース（.get(), .post(), .put(), .delete()など）を提供し、json=やdata=引数でペイロードを指定できます。レスポンスオブジェクトもRequestsのものと似ており、.status_code, .text, .json()などで結果を確認できます。

デプロイ 🚀☁️

ResponderアプリケーションはASGIアプリケーションなので、Uvicorn, Hypercorn, DaphneなどのASGIサーバーを使ってデプロイする必要があります。api.run()は開発用であり、プロダクション環境での使用は推奨されません。

Uvicornを使った基本的なデプロイ

最も一般的な方法はUvicornを使うことです。まずUvicornをインストールします。

pip install uvicorn

そして、app.py（api = responder.API()が含まれるファイル）があるディレクトリで以下のコマンドを実行します。

uvicorn app:api --host 0.0.0.0 --port 8000
  • app:api: app.pyファイル内のapiという名前のASGIアプリケーションオブジェクトを指定します。
  • --host 0.0.0.0: すべてのネットワークインターフェースでリッスンします（外部からのアクセスを許可）。
  • --port 8000: ポート8000でリッスンします。
  • --workers 4: （オプション）ワーカープロセス数を指定してパフォーマンスを向上させます。
  • --reload: （開発時）コードが変更されたら自動的にリロードします。プロダクションでは使用しません。

Gunicorn + Uvicornワーカー

より堅牢なプロセス管理のために、Gunicornをプロセススーパーバイザーとして使用し、Uvicornをワーカークラスとして指定する方法もよく使われます。

pip install gunicorn uvicorn
gunicorn app:api -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000
  • -w 4: ワーカープロセスの数を4つに設定します。CPUコア数などを参考に調整します。
  • -k uvicorn.workers.UvicornWorker: GunicornにUvicornワーカーを使用するよう指示します。
  • --bind 0.0.0.0:8000: リッスンするアドレスとポートを指定します。

Dockerを使ったデプロイ

アプリケーションをコンテナ化してデプロイすることも一般的です。以下は簡単なDockerfileの例です。

# ベースイメージを選択 (Python 3.7以降)
FROM python:3.9-slim

# 作業ディレクトリを設定
WORKDIR /app

# 依存関係ファイルをコピー
COPY Pipfile Pipfile.lock ./
# または requirements.txt をコピー
# COPY requirements.txt ./

# Pipenvを使って依存関係をインストール (システム全体に)
RUN pip install pipenv && pipenv install --system --deploy --ignore-pipfile
# または pip を使う場合
# RUN pip install --no-cache-dir -r requirements.txt

# アプリケーションコードをコピー
COPY . .

# アプリケーションがリッスンするポートを指定
EXPOSE 8000

# コンテナ起動時にUvicornを実行
CMD ["uvicorn", "app:api", "--host", "0.0.0.0", "--port", "8000"]

このDockerfileをビルドし、実行することで、Responderアプリケーションをコンテナとしてデプロイできます。

PaaS (Herokuなど)

HerokuのようなPaaSプラットフォームにもデプロイできます。通常、Procfileを作成してGunicorn + Uvicornの起動コマンドを指定します。

Procfile:

web: gunicorn app:api -w 4 -k uvicorn.workers.UvicornWorker

依存関係はPipfileやrequirements.txtで管理します。

まとめと注意点 📌

Responderは、PythonのWebフレームワークの中でも、特に開発者の体験を重視して設計された、シンプルで使いやすいフレームワークでした。ASGIへの早期対応、直感的なルーティング、組み込みのテストクライアントやWebSocketサポートなど、モダンなWebアプリケーション開発に必要な機能を多く備えていました。

しかし、冒頭でも述べたように、Responderは現在活発に開発されておらず、作者自身もFastAPIの使用を推奨しています。FastAPIはResponderと同様にStarletteをベースにしており、型ヒントを活用したデータ検証や自動ドキュメント生成など、さらに多くの機能と高いパフォーマンスを提供し、コミュニティも非常に活発です。

これから新しいプロジェクトを始める場合、特にAPI開発や非同期処理が重要な場合は、FastAPIやStarlette、Django (バージョン3以降はASGIサポートあり)、Flask (ASGIサポートを追加可能) などを検討することをお勧めします。

Responderは、ASGIフレームワークの初期の設計思想や、Kenneth Reitz氏のAPI設計哲学を学ぶ上で依然として価値がありますが、プロダクション環境での新規採用は慎重に判断する必要があります。このガイドがResponderの理解の一助となれば幸いです 😊。

参考情報 📚

コメント

タイトルとURLをコピーしました