# 付録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__()` が内部で実装しているものと同一です。 ```python # 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"] ``` このファイルを保存して以下のコマンドで起動します。 ```bash 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` の中身を覗いてみましょう。リクエストのメソッドとパスを表示するように拡張します。 ```python # 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` に設定するだけです。 ```python # 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] ``` ```bash 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 のミドルウェアチェーンの基盤にあるのは、この単純なラッピング構造です。 ```python # 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)) ``` ```bash 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 の最小アプリケーションを構築し、非同期の世界での同等の構造を確認します。