付録A 最小 WSGI アプリ集

本書の第1部から第3部にかけて、Django と FastAPI の内部で WSGI と ASGI がどのように機能しているかを解説してきました。しかしフレームワークの内部を追いかけているだけでは、WSGI そのものがどれほどシンプルな規約であるかが見えにくくなることがあります。本付録では、フレームワークを一切使わずに WSGI の仕様だけで動作する最小のアプリケーションを3つ示します。いずれも Gunicorn でそのまま起動でき、ブラウザや curl で動作を確認できます。フレームワークが裏側で行っている仕事を「自分の手で書いてみる」ことで、抽象化の下にある構造を体感してください。

Hello World

WSGI アプリケーションの最小形は、callable（呼び出し可能オブジェクト）が environ と start_response の2つの引数を受け取り、レスポンスボディをイテラブルとして返す、それだけです。 PEP 3333 で定義されたこの規約は、Django の WSGIHandler.__call__() が内部で実装しているものと同一です。

# app_hello.py

def application(environ, start_response):
    status = "200 OK"
    headers = [("Content-Type", "text/plain; charset=utf-8")]
    start_response(status, headers)
    return [b"Hello, WSGI World!\n"]

このファイルを保存して以下のコマンドで起動します。

gunicorn app_hello:application --bind 127.0.0.1:8000

curl http://127.0.0.1:8000/ を実行すると Hello, WSGI World! が返されます。

この7行のコードの中に、WSGI の本質がすべて詰まっています。 environ は辞書で、REQUEST_METHOD, PATH_INFO, QUERY_STRING, HTTP_HOST など、HTTP リクエストに関するすべての情報が格納されています。 start_response はサーバから渡されるコールバック関数で、ステータスコードとレスポンスヘッダを受け取ります。戻り値はバイト列のイテラブルで、これがレスポンスボディになります。

Django の WSGIHandler は、この同じインターフェースの裏側で HttpRequest オブジェクトの構築、ミドルウェアチェーンの実行、URL ルーティング、ビュー関数の呼び出し、HttpResponse から [bytes] への変換を行っています。最小の WSGI アプリを書いてみると、フレームワークがどれだけ多くの仕事を引き受けてくれているかを実感できます。

environ の中身を覗いてみましょう。リクエストのメソッドとパスを表示するように拡張します。

# app_echo.py

def application(environ, start_response):
    method = environ["REQUEST_METHOD"]
    path = environ["PATH_INFO"]
    query = environ.get("QUERY_STRING", "")
    body = f"Method: {method}\nPath: {path}\nQuery: {query}\n"

    status = "200 OK"
    headers = [
        ("Content-Type", "text/plain; charset=utf-8"),
        ("Content-Length", str(len(body.encode("utf-8")))),
    ]
    start_response(status, headers)
    return [body.encode("utf-8")]

curl http://127.0.0.1:8000/users?page=2 を実行すると、メソッド、パス、クエリ文字列がそのまま返されます。 Django のビュー関数で request.method や request.path として参照している値が、environ 辞書のどのキーから来ているかが一目瞭然です。

JSON API

次に、JSON を返す API エンドポイントを WSGI だけで構築します。フレームワークを使えば JsonResponse や Pydantic モデルが JSON のシリアライズを担いますが、WSGI の素朴な世界では json.dumps() でバイト列を生成し、Content-Type を application/json に設定するだけです。

# app_json.py
import json
from urllib.parse import parse_qs

ITEMS = [
    {"id": 1, "name": "Django 本", "price": 3200},
    {"id": 2, "name": "FastAPI 本", "price": 2800},
    {"id": 3, "name": "Python 入門", "price": 2400},
]


def application(environ, start_response):
    method = environ["REQUEST_METHOD"]
    path = environ["PATH_INFO"]

    if method == "GET" and path == "/api/items":
        return handle_list_items(environ, start_response)

    if method == "GET" and path.startswith("/api/items/"):
        return handle_get_item(environ, start_response, path)

    return handle_not_found(environ, start_response)


def handle_list_items(environ, start_response):
    query = parse_qs(environ.get("QUERY_STRING", ""))
    max_price = query.get("max_price", [None])[0]

    results = ITEMS
    if max_price is not None:
        try:
            max_price = int(max_price)
            results = [item for item in ITEMS if item["price"] <= max_price]
        except ValueError:
            return json_response(start_response, "400 Bad Request",
                                 {"error": "max_price must be an integer"})

    return json_response(start_response, "200 OK", {"items": results})


def handle_get_item(environ, start_response, path):
    try:
        item_id = int(path.split("/")[-1])
    except ValueError:
        return json_response(start_response, "400 Bad Request",
                             {"error": "Invalid item ID"})

    item = next((i for i in ITEMS if i["id"] == item_id), None)
    if item is None:
        return json_response(start_response, "404 Not Found",
                             {"error": "Item not found"})

    return json_response(start_response, "200 OK", item)


def handle_not_found(environ, start_response):
    return json_response(start_response, "404 Not Found",
                         {"error": "Not found", "path": environ["PATH_INFO"]})


def json_response(start_response, status, data):
    body = json.dumps(data, ensure_ascii=False).encode("utf-8")
    headers = [
        ("Content-Type", "application/json; charset=utf-8"),
        ("Content-Length", str(len(body))),
    ]
    start_response(status, headers)
    return [body]

gunicorn app_json:application --bind 127.0.0.1:8000

curl http://127.0.0.1:8000/api/items
curl http://127.0.0.1:8000/api/items/2
curl http://127.0.0.1:8000/api/items?max_price=3000
curl http://127.0.0.1:8000/api/items/999
curl http://127.0.0.1:8000/nonexistent

このコードは約60行ですが、その中にはルーティング（パスの分岐）、クエリパラメータの解析、型変換とバリデーション（int() の try/except）、エラーレスポンスの生成、JSON シリアライズといった、フレームワークが通常担う責務がすべて手動で実装されています。

14-4 で学んだ「入力値を信用しない」原則が、ここでも適用されていることに注目してください。 item_id を int() で変換する際に ValueError を捕捉し、不正な入力に対して 400 を返しています。 max_price のクエリパラメータも同様です。フレームワークなしの世界では、この防御を自分で書かなければなりません。 Django のフォームや FastAPI の Pydantic モデルがどれほどの労力を削減しているかが、この比較で明確になります。

middleware 例

最後に、WSGI ミドルウェアの仕組みを示します。 WSGI ミドルウェアは、アプリケーションをラップする callable であり、リクエストの前処理やレスポンスの後処理を行います。 Django のミドルウェアチェーンの基盤にあるのは、この単純なラッピング構造です。

# app_middleware.py
import json
import time
import logging

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")
logger = logging.getLogger(__name__)


# ---- アプリケーション本体 ----

def app(environ, start_response):
    path = environ["PATH_INFO"]
    if path == "/api/health":
        body = json.dumps({"status": "ok"}).encode("utf-8")
        start_response("200 OK", [
            ("Content-Type", "application/json"),
            ("Content-Length", str(len(body))),
        ])
        return [body]

    start_response("404 Not Found", [("Content-Type", "text/plain")])
    return [b"Not Found\n"]


# ---- ミドルウェア 1: リクエストログ ----

class RequestLoggingMiddleware:
    def __init__(self, app):
        self.app = app

    def __call__(self, environ, start_response):
        start = time.time()
        method = environ["REQUEST_METHOD"]
        path = environ["PATH_INFO"]

        captured_status = [None]

        def custom_start_response(status, headers, exc_info=None):
            captured_status[0] = status
            return start_response(status, headers, exc_info)

        response = self.app(environ, custom_start_response)
        duration = time.time() - start
        logger.info(f'{method} {path} {captured_status[0]} {duration:.4f}s')
        return response


# ---- ミドルウェア 2: セキュリティヘッダ付与 ----

class SecurityHeadersMiddleware:
    def __init__(self, app):
        self.app = app

    def __call__(self, environ, start_response):
        def custom_start_response(status, headers, exc_info=None):
            headers.append(("X-Content-Type-Options", "nosniff"))
            headers.append(("X-Frame-Options", "DENY"))
            headers.append(("Referrer-Policy", "strict-origin-when-cross-origin"))
            return start_response(status, headers, exc_info)

        return self.app(environ, custom_start_response)


# ---- ミドルウェアの適用（内側から外側へラップ） ----

application = RequestLoggingMiddleware(SecurityHeadersMiddleware(app))

gunicorn app_middleware:application --bind 127.0.0.1:8000

curl -I http://127.0.0.1:8000/api/health

curl -I の出力には、アプリケーション本体が設定した Content-Type: application/json に加えて、SecurityHeadersMiddleware が追加した X-Content-Type-Options: nosniff, X-Frame-Options: DENY, Referrer-Policy: strict-origin-when-cross-origin が含まれています。ターミナルにはリクエストログミドルウェアが出力した GET /api/health 200 OK 0.0002s のようなログが表示されます。

このコードで注目すべきは、ミドルウェアの適用順序です。最終行の application = RequestLoggingMiddleware(SecurityHeadersMiddleware(app)) は、リクエストがまず RequestLoggingMiddleware に入り、次に SecurityHeadersMiddleware を通過し、最後に app に到達することを意味します。レスポンスは逆順に戻ります。 15-2 で「Django のミドルウェアはリクエスト時に上から下へ、レスポンス時に下から上へ処理される」と説明したのは、まさにこのラッピング構造のことです。

custom_start_response を使ってレスポンスのステータスコードやヘッダを横取りする手法は、Django の process_response や FastAPI の @app.middleware("http") が内部で行っている操作の素朴な形です。フレームワークはこの仕組みをより洗練された API で包んでくれていますが、根底にある原理は同じです。

以上の3つのアプリケーションは、WSGI という規約が驚くほどシンプルであることを示しています。 callable が environ を受け取り、start_response を呼び、バイト列を返す。この約束事だけで、Hello World から JSON API、ミドルウェアチェーンまで構築できます。 Django と FastAPI の内部構造を学ぶ旅は、最終的にこのシンプルな規約に帰着します。付録 B では、同じアプローチで ASGI の最小アプリケーションを構築し、非同期の世界での同等の構造を確認します。