(Django は ASGI にどう対応しているか)= # Django は ASGI にどう対応しているか ## asgi.py Django プロジェクトを `django-admin startproject myproject` で生成すると、`wsgi.py` と並んで `asgi.py` が自動的に作成されます。 ```python # myproject/asgi.py import os from django.core.asgi import get_asgi_application os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings') application = get_asgi_application() ``` この構造は{numref}`Django を WSGI 視点で見る`（{ref}`Django を WSGI 視点で見る`）で見た `wsgi.py` とほぼ同一です。 ```python # myproject/wsgi.py（比較） import os from django.core.wsgi import get_wsgi_application os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings') application = get_wsgi_application() ``` 2つのファイルの違いは非常にシンプルです。 | 項目 | wsgi.py | asgi.py | |------|---------|---------| | インポート元 | `django.core.wsgi` | `django.core.asgi` | | 関数名 | `get_wsgi_application` | `get_asgi_application` | | 起動コマンド | `gunicorn myproject.wsgi:application` | `uvicorn myproject.asgi:application` | | インタフェース | `application(environ, start_response)` | `await application(scope, receive, send)` | どちらも `DJANGO_SETTINGS_MODULE` 環境変数の設定と `django.setup()` の呼び出し（アプリレジストリの初期化）を行い、WSGI/ASGI 準拠の `application` callable を返します。 本番環境では同一プロジェクトが WSGI と ASGI の両方のエントリーポイントを持つことになりますが、実際に使用するのはどちらか一方です。 - ASGI サーバで起動する場合、`wsgi.py` は使われません - ただし両ファイルを残しておくことで、ASGI サーバに問題が生じた場合に WSGI へ切り戻す選択肢を維持できます ```{tip} 本番環境では `wsgi.py` と `asgi.py` の両方を残しておくことをお勧めします。ASGI サーバに問題が発生したとき、`gunicorn myproject.wsgi:application` に切り替えるだけで WSGI での運用に戻せます。 ``` ## get_asgi_application() `get_asgi_application()` の内部を追うと、Django の ASGI 対応がどのように実装されているかが見えてきます。 ```python # django/core/asgi.py（概念コード） import django from django.core.handlers.asgi import ASGIHandler def get_asgi_application(): django.setup(set_prefix=False) return ASGIHandler() ``` `django.setup()` は以下の処理を行います。 - `INSTALLED_APPS` に登録されたアプリのロード - モデルの初期化 - シグナルの接続 この処理は `get_wsgi_application()` と完全に同一です。返されるのは `ASGIHandler` のインスタンスで、これが ASGI の `application` callable として機能します。 `ASGIHandler` は `django/core/handlers/asgi.py` に定義されており、{numref}`なぜ ASGI が必要になったのか`（{ref}`なぜ ASGI が必要になったのか`）で学んだ ASGI の3引数インタフェースを実装しています。 ```python # django/core/handlers/asgi.py（概念的な構造） class ASGIHandler(base.BaseHandler): """Django の ASGI リクエストハンドラ""" async def __call__(self, scope, receive, send): if scope["type"] == "http": await self.handle(scope, receive, send) elif scope["type"] == "lifespan": # Django 5.0+ では lifespan プロトコルに対応 while True: message = await receive() if message["type"] == "lifespan.startup": await send({"type": "lifespan.startup.complete"}) elif message["type"] == "lifespan.shutdown": await send({"type": "lifespan.shutdown.complete"}) return else: raise ValueError(f"Unknown scope type: {scope['type']}") async def handle(self, scope, receive, send): request = ASGIRequest(scope, await self.read_body(receive)) response = await self.get_response(request) await response(scope, receive, send) async def read_body(self, receive): body = b"" while True: message = await receive() body += message.get("body", b"") if not message.get("more_body", False): break return body ``` ここで注目すべき点がいくつかあります。 **1. `scope["type"]` による分岐** {numref}`ASGI の基本構造`（{ref}`ASGI の基本構造`）で学んだ通り、ASGI アプリケーションは `http`、`websocket`、`lifespan` の3種のスコープを処理する必要があります。Django の `ASGIHandler` は `http` と `lifespan` に対応しますが、`websocket` は標準では対応していません。WebSocket を扱う場合は Django Channels を導入します。 **2. `read_body` メソッド** {numref}`request body を受け取る`（{ref}`request body を受け取る`）で手書きした `read_body(receive)` ヘルパーと本質的に同じ処理が Django 内部にも存在しており、`more_body` フラグを監視しながらループして全ボディを蓄積します。 **3. `ASGIRequest` の生成** {numref}`リクエストは Django にどう渡るか`（{ref}`リクエストは Django にどう渡るか`）で `WSGIRequest` が `environ` 辞書からリクエスト属性を構築したように、`ASGIRequest` は `scope` 辞書と受信済みボディからリクエスト属性を構築します。 ```python # django/core/handlers/asgi.py（概念コード） class ASGIRequest(HttpRequest): def __init__(self, scope, body): self.scope = scope self._stream = BytesIO(body) self.method = scope["method"] self.path_info = scope.get("path", "/") self.META = self._build_meta(scope) def _build_meta(self, scope): meta = { "REQUEST_METHOD": scope["method"], "QUERY_STRING": scope.get("query_string", b"").decode("latin-1"), "SCRIPT_NAME": scope.get("root_path", ""), "PATH_INFO": scope.get("path", "/"), } if scope.get("server"): meta["SERVER_NAME"] = scope["server"][0] meta["SERVER_PORT"] = str(scope["server"][1]) if scope.get("client"): meta["REMOTE_ADDR"] = scope["client"][0] for header_name, header_value in scope.get("headers", []): name = header_name.decode("latin-1") value = header_value.decode("latin-1") if name == "content-type": meta["CONTENT_TYPE"] = value elif name == "content-length": meta["CONTENT_LENGTH"] = value else: key = f"HTTP_{name.upper().replace('-', '_')}" meta[key] = value return meta ``` `_build_meta` の処理は、{numref}`scope を理解する`（{ref}`scope を理解する`）で見た `scope` のキーを{numref}`リクエストは Django にどう渡るか`（{ref}`リクエストは Django にどう渡るか`）で見た `environ` のキー形式（`HTTP_*` プレフィックス付き大文字）に逆変換しています。Django の内部コード（ミドルウェア、認証、CSRF 検証など）はすべて `request.META["HTTP_HOST"]` や `request.META["CONTENT_TYPE"]` の形式を前提としているため、ASGI の `scope["headers"]`（小文字バイト列タプル）をこの形式に合わせる必要があります。 ```{note} この変換層が存在することで、Django のビュー関数、ミドルウェア、テンプレートシステムは WSGI と ASGI のどちらで動作しているかを意識する必要がありません。`request.method`、`request.GET`、`request.POST`、`request.headers`、`request.user` のすべてが同じように動作します。 ``` 入力形式を比較すると次のようになります。 | 入力 | 変換元 | 変換先 | |------|--------|--------| | WSGI の場合 | `environ (dict)` | `WSGIRequest → request.META, request.method, ...` | | ASGI の場合 | `scope (dict) + body (bytes)` | `ASGIRequest → request.META, request.method, ...` | **4. レスポンスの送信** `self.get_response(request)` は{numref}`middleware chain の流れ`（{ref}`middleware chain の流れ`）で見たミドルウェアチェーンを通じてビュー関数を実行し、`HttpResponse` を返します。このレスポンスオブジェクトに対して `await response(scope, receive, send)` を呼ぶことで、ASGI のイベント形式に変換されます。 ```python # django/http/response.py（ASGI 対応部分の概念コード） class HttpResponseBase: async def __call__(self, scope, receive, send): await send({ "type": "http.response.start", "status": self.status_code, "headers": [(k.encode(), v.encode()) for k, v in self.headers.items()], }) # StreamingHttpResponse の場合はチャンク送信 if hasattr(self, "streaming_content"): for chunk in self.streaming_content: await send({"type": "http.response.body", "body": chunk, "more_body": True}) await send({"type": "http.response.body", "body": b"", "more_body": False}) else: await send({"type": "http.response.body", "body": self.content}) ``` {numref}`最小の ASGI HTTP アプリ`（{ref}`最小の ASGI HTTP アプリ`）で手書きした `send({"type": "http.response.start", ...})` と `send({"type": "http.response.body", ...})` の2段階送信が、Django の `HttpResponse` にも組み込まれています。{numref}`ch06-レスポンス生成`（{ref}`ch06-レスポンス生成`）で見た `StreamingHttpResponse` は `more_body=True` でチャンク送信を行い、通常の `HttpResponse` は全コンテンツを一度に送ります。 全体のリクエスト・レスポンスフローをまとめると次のようになります。 ``` Uvicorn / Daphne │ await application(scope, receive, send) ▼ ASGIHandler.__call__ │ scope["type"] == "http" ▼ ASGIHandler.handle ├─ body = await read_body(receive) ← 第8-2節の手書き実装と同等 ├─ request = ASGIRequest(scope, body) ← scope → environ 形式に変換 ├─ response = await get_response(request) ← ミドルウェア → URL解決 → ビュー実行 │ （第6-3〜6-7節と同一のフロー） └─ await response(scope, receive, send) ← HttpResponse → ASGI イベントに変換 ├─ send(http.response.start) └─ send(http.response.body) ``` ```{important} `ASGIHandler.handle` から `get_response` が呼ばれた後は、{numref}`Django を WSGI 視点で見る`（{ref}`Django を WSGI 視点で見る`）で追った WSGI 時のフロー（ミドルウェアチェーン、URL 解決、ビュー実行、レスポンス生成）がまったく同じように動作します。ASGI 対応によって変わったのは入口（`scope` / `receive` → `ASGIRequest`）と出口（`HttpResponse` → `send` イベント）の変換層だけであり、Django の中核ロジックは WSGI / ASGI に対して透過的に設計されています。 ``` ```{mermaid} flowchart LR UV["Uvicorn
await app(scope, receive, send)"] UV --> AH["ASGIHandler.__call__"] AH --> RB["read_body(receive)
ボディ全蓄積"] RB --> AR["ASGIRequest(scope, body)
scope → environ 形式変換"] AR --> GR["get_response(request)
ミドルウェア → URL解決 → ビュー"] GR --> RS["HttpResponse.__call__
send http.response.start
send http.response.body"] ``` 次節では、Django の `async def` ビューの仕組みと、同期ビューが ASGI 上でどのように実行されるかを詳しく追います。 (Django の request 処理と async)= ## Django の request 処理と async ### 同期 view Django の伝統的なビュー関数は、同期関数として定義されます。 ```python from django.http import JsonResponse from django.contrib.auth.decorators import login_required @login_required def user_detail(request, user_id): user = User.objects.get(id=user_id) # 同期 ORM クエリ return JsonResponse({"id": user.id, "name": user.name}) ``` WSGI サーバ（Gunicorn）上でこのビューが実行される場合、すべてがシンプルです。リクエストを受け取ったワーカースレッドがビュー関数を直接呼び出し、ORM クエリが完了するまでそのスレッドがブロックされ、レスポンスを返します。{numref}`Django を WSGI 視点で見る`（{ref}`Django を WSGI 視点で見る`）で追った WSGI フローそのものです。 問題は、ASGI サーバ（Uvicorn / Daphne）上で同期ビューが呼ばれた場合です。 - ASGI サーバはイベントループ上で `await ASGIHandler.__call__(scope, receive, send)` を実行しています - もし同期ビューをイベントループ上で直接呼び出すと、`User.objects.get()` がデータベース応答を待つ間、イベントループ全体がブロックされます - これは {numref}`ch09-トラブルシューティングの観点`（{ref}`ch09-トラブルシューティングの観点`）で見た「`async def` 内での blocking I/O」と同じ問題です Django はこの問題を自動的に解決します。`ASGIHandler` はビュー関数が同期関数かどうかを `asyncio.iscoroutinefunction()` で検査し、同期関数であればスレッドプールにオフロードして実行します。 ```python # django/core/handlers/base.py（概念コード） class BaseHandler: async def get_response_async(self, request): response = None callback, callback_args, callback_kwargs = self.resolve_request(request) # ビュー関数が同期か非同期かを判定 if asyncio.iscoroutinefunction(callback): response = await callback(request, *callback_args, **callback_kwargs) else: # 同期関数をスレッドプールで実行 response = await asyncio.to_thread( callback, request, *callback_args, **callback_kwargs ) return response ``` `asyncio.to_thread()` は関数をデフォルトのスレッドプールエグゼキュータで実行し、完了を `await` で待ちます。スレッドプール内で同期ビューが `User.objects.get()` でブロックしている間も、イベントループは他のリクエストを処理できます。これは FastAPI が `def`（同期）エンドポイントを `anyio.to_thread.run_sync` で実行するのと同じ設計判断です（{numref}`FastAPI を ASGI 視点で見る`（{ref}`FastAPI を ASGI 視点で見る`））。 ミドルウェアについても同様の処理が行われます。Django のミドルウェアは歴史的に同期で書かれているものが大半です。`ASGIHandler` はミドルウェアチェーンの各ミドルウェアが同期か非同期かを判定し、同期ミドルウェアはスレッドプールで実行します。この判定と切り替えの処理が、ASGI 上で既存の同期コードをそのまま動作させる互換層として機能しています。 ### 非同期 view Django 4.1 以降では、ビュー関数を `async def` で定義できるようになりました。 ```python from django.http import JsonResponse async def user_detail(request, user_id): user = await User.objects.aget(id=user_id) # 非同期 ORM クエリ return JsonResponse({"id": user.id, "name": user.name}) ``` `async def` で定義されたビューは、ASGI サーバのイベントループ上で直接実行されます。スレッドプールへのオフロードは行われず、`await` で I/O 待ちをしている間にイベントループが他のリクエストを処理できます。 Django の ORM は従来すべて同期でしたが、4.1 以降で非同期インタフェースが段階的に追加されています。`QuerySet` の主要なメソッドに `a` プレフィックスの非同期版が用意されています。 | 同期 ORM | 非同期 ORM | |----------|-----------| | `User.objects.get(id=1)` | `await User.objects.aget(id=1)` | | `User.objects.filter(active=True)` | `User.objects.filter(active=True)`（QuerySet 構築は同期） | | `list(queryset)` | `async for user in queryset:`（評価が非同期） | | `User.objects.create(name="Taro")` | `await User.objects.acreate(name="Taro")` | | `user.save()` | `await user.asave()` | | `user.delete()` | `await user.adelete()` | | `queryset.count()` | `await queryset.acount()` | | `queryset.exists()` | `await queryset.aexists()` | 注意すべきは、非同期 ORM メソッドの内部実装です。Django 4.x 時点では、多くの非同期 ORM メソッドは内部的に `sync_to_async` を使って同期版のメソッドをスレッドプールで実行しています。つまり `await User.objects.aget(id=1)` は、概念的には `await asyncio.to_thread(User.objects.get, id=1)` に近い動作をしています。データベースドライバ自体が非同期（`asyncpg` など）になるわけではなく、同期ドライバ（`psycopg2` など）をスレッドプールで呼び出す形です。 ```python # Django ORM の非同期メソッドの内部（概念コード） class QuerySet: async def aget(self, **kwargs): return await sync_to_async(self.get)(**kwargs) ``` ```{note} 「非同期ビューを書いてもデータベースアクセスが本質的に非同期になるわけではない」という点に注意してください。パフォーマンス上の利点は、イベントループがブロックされないことにより他のリクエストを並行処理できる点にあります。真の非同期データベースアクセスが必要な場合は、`encode/databases` や SQLAlchemy の async 拡張を Django と組み合わせることになります。 ``` 非同期ビュー内で同期 ORM メソッドを直接呼ぶと、Django は `SynchronousOnlyOperation` 例外を送出します。 ```python async def bad_view(request, user_id): user = User.objects.get(id=user_id) # SynchronousOnlyOperation! return JsonResponse({"id": user.id}) ``` ```{important} この例外は安全弁として機能しています。`async def` 内で同期ブロッキング呼び出しが行われることを検出し、明示的にエラーにすることで、{numref}`ch09-トラブルシューティングの観点`（{ref}`ch09-トラブルシューティングの観点`）で述べたイベントループブロッキングの問題を未然に防ぎます。FastAPI では同じ問題が警告なしに発生するのに対し、Django はフレームワークレベルで検出する仕組みを持っています。 ``` `sync_to_async` デコレータを使えば、任意の同期関数を非同期ビュー内から安全に呼び出せます。 ```python from asgiref.sync import sync_to_async @sync_to_async def get_user_sync(user_id): return User.objects.select_related("profile").get(id=user_id) async def user_detail(request, user_id): user = await get_user_sync(user_id) return JsonResponse({"id": user.id, "name": user.name}) ``` 逆に、同期コードから非同期関数を呼びたい場合は `async_to_sync` を使います。 ```{tip} `sync_to_async` と `async_to_sync` の使い分けをまとめると、「同期 → 非同期方向」は `sync_to_async`、「非同期 → 同期方向」は `async_to_sync` です。 ``` ```python from asgiref.sync import async_to_sync def sync_view(request): result = async_to_sync(fetch_external_api)("https://api.example.com/data") return JsonResponse(result) ``` ### 実行経路の違い 同期ビューと非同期ビューの実行経路を並べると、Django が ASGI 上で2つの異なるパスを持っていることがわかります。 ```{mermaid} flowchart TD AH["ASGIHandler
await __call__"] AH --> IC{"iscoroutinefunction
view?"} IC -->|async def| AV["イベントループ上で
await view(request)"] IC -->|def| SV["asyncio.to_thread
スレッドプールで実行"] AV --> AO["await aget() など
非同期 ORM"] AV --> SE["同期 ORM 呼び出し
→ SynchronousOnlyOperation"] SV --> SO["通常の同期 ORM
そのまま使用可"] ``` ``` ASGI サーバ (Uvicorn) │ await ASGIHandler.__call__(scope, receive, send) │ ├─ body = await read_body(receive) ├─ request = ASGIRequest(scope, body) │ ├─ ミドルウェアチェーン │ ├─ 同期ミドルウェア → sync_to_async でスレッドプール実行 │ └─ 非同期ミドルウェア → イベントループ上で直接実行 │ ├─ URL 解決 → ビュー関数を特定 │ ├─ iscoroutinefunction(view) で判定 │ │ │ ├─ True（async def view） │ │ → イベントループ上で直接 await view(request, ...) │ │ → await aget() / async for ... で非同期 ORM │ │ → 同期呼び出しは SynchronousOnlyOperation で拒否 │ │ │ └─ False（def view） │ → asyncio.to_thread(view, request, ...) でスレッドプール実行 │ → 通常の同期 ORM がそのまま使える │ → イベントループはブロックされない │ ├─ response = HttpResponse / JsonResponse └─ await response(scope, receive, send) ├─ send(http.response.start) └─ send(http.response.body) ``` WSGI サーバ上で動作する場合は、この分岐がまったく異なります。 ``` WSGI サーバ (Gunicorn) │ WSGIHandler.__call__(environ, start_response) │ ├─ request = WSGIRequest(environ) │ ├─ ミドルウェアチェーン（すべて同期実行） │ ├─ URL 解決 → ビュー関数を特定 │ ├─ iscoroutinefunction(view) で判定 │ │ │ ├─ True（async def view） │ │ → async_to_sync(view)(request, ...) でイベントループを作成して実行 │ │ → 動作はするが非同期の利点がない │ │ │ └─ False（def view） │ → view(request, ...) を直接呼び出し │ ├─ response = HttpResponse └─ start_response(status, headers) + return response ``` WSGI サーバ上で `async def` ビューを実行した場合、Django は `async_to_sync` を使って同期的に実行します。動作はしますが、イベントループが新たに作られて即座にブロックされるため、非同期の利点（I/O 待ち中の並行処理）は得られません。`async def` ビューの真価は ASGI サーバ上でのみ発揮されます。 この設計判断は Django の段階的移行戦略を反映しています。 - 既存の同期コード（ビュー、ミドルウェア、ORM）は ASGI 上でもスレッドプール経由で動作し、破壊的変更なしに移行できます - 新しいコードは `async def` で書くことで非同期の利点を得られ、プロジェクト内で同期と非同期のビューを混在させることも可能です - FastAPI が最初から ASGI ネイティブで設計されたのに対し、Django は14年以上の歴史を持つ同期コードベースとの互換性を維持しながら非同期対応を進めています 実務上の判断基準は次の通りです。 - **`async def` を選ぶべき場合**: 外部 API の呼び出しやファイル I/O が多いビュー - **`def` のまま維持すべき場合**: ORM ヘビーなビュー ```{caution} `sync_to_async` と `async_to_sync` のブリッジは便利ですが、呼び出しのたびにスレッドプールのオーバーヘッドが発生します。頻繁な切り替えはパフォーマンスに影響するため、境界を越える回数を最小限に抑えることを意識してください。 ``` 次節では Django Channels を通じた WebSocket 対応と、Django の ASGI 拡張の全体像を確認します。 (sync / async 境界の橋渡し)= ## sync / async 境界の橋渡し ### asgiref Django の同期・非同期変換は `asgiref` というライブラリが担っています。`asgiref` は Django プロジェクトが開発・保守しているパッケージで、Django をインストールすると依存関係として自動的にインストールされます。 `asgiref` は ASGI 仕様のリファレンス実装とユーティリティを提供するライブラリですが、その中で最も重要な機能が `sync_to_async` と `async_to_sync` の2つのブリッジ関数です。これらは Django 内部のあらゆる場所で使われており、{numref}`Django の request 処理と async`（{ref}`Django の request 処理と async`）で見た「同期ビューを ASGI 上でスレッドプール実行する」仕組みも、「非同期ビューを WSGI 上で実行する」仕組みも、すべて `asgiref` を通じて実現されています。 ```python # asgiref は Django と一緒にインストール済み from asgiref.sync import sync_to_async, async_to_sync ``` `asgiref` が解決する問題を理解するには、Python の `asyncio` における基本的な制約を知る必要があります。 - `asyncio` のイベントループは単一スレッドで動作し、`await` 可能なコルーチンだけを扱います - 同期関数をイベントループ上で直接呼ぶとブロッキングが発生します（{numref}`ch07-トラブルシューティングの観点`（{ref}`ch07-トラブルシューティングの観点`）、{numref}`ch09-トラブルシューティングの観点`（{ref}`ch09-トラブルシューティングの観点`）） - 逆にイベントループが走っていない同期コンテキストではコルーチンを `await` できません `asgiref` はこの双方向の壁を、スレッドプールとイベントループの生成・管理によって透過的に越える仕組みを提供しています。 ### sync_to_async `sync_to_async` は、同期関数を非同期コンテキストから安全に呼び出すためのラッパーです。 ```python from asgiref.sync import sync_to_async # 同期関数 def get_user_from_db(user_id): return User.objects.select_related("profile").get(id=user_id) # async def ビュー内から呼ぶ async def user_detail(request, user_id): get_user = sync_to_async(get_user_from_db) user = await get_user(user_id) return JsonResponse({"id": user.id, "name": user.name}) ``` `sync_to_async` が返すのは、元の同期関数をスレッドプールで実行し、その結果を `await` で受け取れるコルーチン関数です。内部的には `asyncio.get_event_loop().run_in_executor()` に近い動作をしますが、`asgiref` はそれに加えて Django 固有の問題を解決する追加機能を持っています。 最も重要な追加機能が `thread_sensitive` パラメータです。 ```python # thread_sensitive=True（デフォルト） @sync_to_async(thread_sensitive=True) def access_orm(): return User.objects.all().count() # thread_sensitive=False @sync_to_async(thread_sensitive=False) def compute_hash(data): return hashlib.sha256(data).hexdigest() ``` 2つのモードの違いをまとめます。 | パラメータ | 実行先 | 用途 | |-----------|--------|------| | `thread_sensitive=True`（デフォルト） | 単一の共有スレッド | Django ORM・データベース操作 | | `thread_sensitive=False` | 通常のスレッドプール（並行実行） | 純粋な計算処理・ファイル I/O | `thread_sensitive=True`（デフォルト）の場合、関数は「メインスレッド」に相当する単一の共有スレッドで実行されます。Django の ORM はスレッドローカルなデータベース接続を使用しており、異なるスレッドで ORM クエリが実行されると、接続の共有やトランザクションの一貫性に問題が生じる可能性があります。`thread_sensitive=True` はこれを防ぐために、同じスレッドで順番に実行されることを保証します。 `thread_sensitive=False` の場合は通常のスレッドプールで実行されます。複数の呼び出しが並行して異なるスレッドで動作するため、データベースに依存しない純粋な計算処理やファイル I/O に適しています。 ```python async def process_data(request): # thread_sensitive=True: 同じスレッドで順次実行 users = await sync_to_async(User.objects.all().count)() # thread_sensitive=False: 別々のスレッドで並行実行可能 hash_a = sync_to_async(compute_hash, thread_sensitive=False) hash_b = sync_to_async(compute_hash, thread_sensitive=False) result_a, result_b = await asyncio.gather( hash_a(b"data_a"), hash_b(b"data_b"), ) return JsonResponse({"users": users, "hash_a": result_a, "hash_b": result_b}) ``` この例で `compute_hash` は `thread_sensitive=False` で並行実行され、2つのハッシュ計算が同時に進みます。一方で ORM クエリは `thread_sensitive=True` で保護されています。 Django の内部では、同期ミドルウェアやビューを ASGI 上で実行する際にも `sync_to_async(thread_sensitive=True)` が使われています。{numref}`Django の request 処理と async`（{ref}`Django の request 処理と async`）で示した実行経路の図で「スレッドプール実行」としていた部分の正体がこれです。 ```python # Django 内部での使用例（概念コード） # django/core/handlers/base.py if not asyncio.iscoroutinefunction(middleware): middleware = sync_to_async(middleware, thread_sensitive=True) ``` デコレータとしても使えます。 ```python @sync_to_async def get_user_permissions(user): return list(user.get_all_permissions()) async def check_permissions(request): permissions = await get_user_permissions(request.user) return JsonResponse({"permissions": permissions}) ``` ### async_to_sync `async_to_sync` は反対方向の変換で、非同期関数を同期コンテキストから呼び出せるようにするラッパーです。 ```python from asgiref.sync import async_to_sync import httpx async def fetch_external_api(url): async with httpx.AsyncClient() as client: response = await client.get(url, timeout=10) return response.json() # 同期ビュー内から呼ぶ def dashboard_view(request): fetch = async_to_sync(fetch_external_api) data = fetch("https://api.example.com/stats") return JsonResponse(data) ``` `async_to_sync` は以下のように動作します。 - イベントループが存在しない場合: 新しいイベントループを作成してコルーチンを実行し、結果を同期的に返します - イベントループが既に存在する場合（例えば ASGI サーバ上の同期ビュー内）: 別スレッドでイベントループを起動して実行します Django の WSGI ハンドラが `async def` ビューを処理する際にも `async_to_sync` が使われています。 ```python # django/core/handlers/base.py（概念コード） if asyncio.iscoroutinefunction(callback): # WSGI 上で async ビューを実行する場合 callback = async_to_sync(callback) response = callback(request, *args, **kwargs) ``` {numref}`Django の request 処理と async`（{ref}`Django の request 処理と async`）で「WSGI サーバ上で `async def` ビューを実行した場合、動作はするが非同期の利点がない」と述べた理由がここにあります。`async_to_sync` はコルーチンを実行完了まで同期的にブロックするため、`await` による並行処理の恩恵は得られません。 `async_to_sync` が最も有用なのは、Django の管理コマンドやテスト、Celery タスクなど、同期的な実行環境から非同期ライブラリを利用する場合です。 ```python # management command from django.core.management.base import BaseCommand from asgiref.sync import async_to_sync class Command(BaseCommand): def handle(self, *args, **options): result = async_to_sync(self.fetch_and_update)() self.stdout.write(f"Updated {result} records") async def fetch_and_update(self): async with httpx.AsyncClient() as client: data = await client.get("https://api.example.com/data") count = await sync_to_async(self.bulk_update)(data.json()) return count ``` この例では `async_to_sync` と `sync_to_async` が組み合わされています。管理コマンド（同期）→ `async_to_sync` で非同期関数を呼び出し → その中で `sync_to_async` で ORM 操作を行う、という流れです。 同期と非同期の境界を越える際には、以下の点に注意が必要です。 **パフォーマンスのオーバーヘッド** `sync_to_async` はスレッドの取得・解放、`async_to_sync` はイベントループの作成・管理にコストがかかります。ループ内で毎回変換を繰り返すと顕著な遅延が生じます。 ```python # 悪い例: ループ内で毎回 sync_to_async async def bad_pattern(request): result = [] for user_id in range(100): user = await sync_to_async(User.objects.get)(id=user_id) # 100回のスレッド切り替え result.append(user.name) return JsonResponse({"users": result}) # 良い例: バッチで取得して1回の変換 async def good_pattern(request): @sync_to_async def get_all_users(): return list(User.objects.filter(id__in=range(100)).values_list("name", flat=True)) names = await get_all_users() # 1回のスレッド切り替え return JsonResponse({"users": names}) ``` **`thread_sensitive=True` のデッドロックリスク** `thread_sensitive=True` の関数は単一スレッドで順次実行されるため、その中からさらに `async_to_sync` を使ってイベントループに戻り、再度 `sync_to_async(thread_sensitive=True)` を呼ぶと、単一スレッドが既に占有されているためデッドロックします。 ```text # デッドロックの可能性がある例 @sync_to_async(thread_sensitive=True) def outer(): # この中で再び async → sync の変換を行うと危険 result = async_to_sync(inner)() return result async def inner(): # thread_sensitive=True のスレッドは outer が占有中 return await sync_to_async(User.objects.count, thread_sensitive=True)() ``` **コンテキストの伝播の問題** `sync_to_async` で実行される関数はメインスレッドとは異なるコンテキストで動作する場合があり、`threading.local()` に保存されたデータ（データベース接続など）が共有されない可能性があります。`thread_sensitive=True` がデフォルトである理由の一つがこれです。 全体の関係を図にすると次のようになります。 ```{mermaid} flowchart LR subgraph 同期の世界 WH["WSGI ハンドラ"] SV["同期ビュー
def view"] ORM["同期 ORM
.get .filter"] CC["Celery / 管理コマンド"] end subgraph 非同期の世界 AH["ASGI ハンドラ"] AV["非同期ビュー
async def view"] AORM["非同期 ORM
.aget async for"] EXT["外部 API
httpx"] end 同期の世界 -->|sync_to_async| 非同期の世界 非同期の世界 -->|async_to_sync| 同期の世界 ``` ``` 同期の世界 非同期の世界 ────────────── ────────────── WSGI ハンドラ ASGI ハンドラ 同期ビュー (def) 非同期ビュー (async def) 同期ミドルウェア 非同期ミドルウェア 同期 ORM (.get, .filter) 非同期 ORM (.aget, async for) Celery タスク 外部 API (httpx) 管理コマンド WebSocket ハンドラ sync_to_async →→→→→→→→→→→→→ ←←←←←←←←←←←←← async_to_sync asgiref が双方向のブリッジを提供 ``` `asgiref` の同期・非同期ブリッジは、Django が14年以上の同期コードベースを維持しながら ASGI の非同期世界に段階的に移行するための中核技術です。FastAPI のように最初から非同期前提で設計されたフレームワークにはこの複雑さが存在しませんが、Django の既存エコシステム（数千のサードパーティパッケージ、同期 ORM、同期ミドルウェア）を活かしつつ非同期の利点を取り入れるために、この橋渡し層は不可欠です。 次節では Django Channels を通じた WebSocket 対応と、Django の ASGI 拡張の全体像を確認します。 (middleware は async にどう対応するか)= ## middleware は async にどう対応するか ### sync middleware Django のミドルウェアは、歴史的にすべて同期で書かれてきました。{numref}`middleware chain の流れ`（{ref}`middleware chain の流れ`）で見た標準的なミドルウェアの構造を振り返ります。 ```python class TimingMiddleware: def __init__(self, get_response): self.get_response = get_response def __call__(self, request): import time start = time.time() response = self.get_response(request) elapsed = (time.time() - start) * 1000 response["X-Process-Time"] = f"{elapsed:.1f}ms" return response ``` WSGI サーバ上では、この `__call__` は同期的に呼ばれ、`self.get_response(request)` が次のミドルウェアまたはビュー関数を同期的に実行し、レスポンスが返ってきます。すべてが単一スレッド上で直線的に進行します。 ASGI サーバ上で同じミドルウェアが使われるとき、Django の `ASGIHandler` はミドルウェアチェーンを構築する段階で各ミドルウェアが同期か非同期かを検査します。同期ミドルウェアが検出された場合、{numref}`sync / async 境界の橋渡し`（{ref}`sync / async 境界の橋渡し`）で見た `sync_to_async` を使って非同期呼び出し可能な形にラップします。 ```python # django/core/handlers/base.py（概念コード） def load_middleware(self, is_async=False): handler = self._get_response_async if is_async else self._get_response for middleware_path in reversed(settings.MIDDLEWARE): middleware_cls = import_string(middleware_path) middleware_instance = middleware_cls(handler) if is_async: # ASGI モードの場合 if asyncio.iscoroutinefunction(middleware_instance.__call__): handler = middleware_instance # そのまま使う else: # 同期ミドルウェアを async でラップ handler = sync_to_async( middleware_instance.__call__, thread_sensitive=True ) else: handler = middleware_instance self._middleware_chain = handler ``` この自動変換により、`settings.py` の `MIDDLEWARE` リストに登録された同期ミドルウェアは ASGI 上でも動作します。開発者が既存のミドルウェアコードを書き換える必要はありません。Django のサードパーティパッケージが提供するミドルウェア（`django-cors-headers`、`django-debug-toolbar` など）も、同期で書かれていればこの変換によって自動的に ASGI 互換になります。 ```{note} {numref}`sync / async 境界の橋渡し`（{ref}`sync / async 境界の橋渡し`）で述べた通り、`sync_to_async(thread_sensitive=True)` はすべてのリクエストで同じ単一スレッドを使います。ミドルウェアチェーン内の同期ミドルウェアが多いほど、そのスレッドがボトルネックになります。リクエスト処理のうちミドルウェア通過部分だけが直列化され、非同期の並行処理の利点が減少します。 ``` ### async middleware Django 4.1 以降では、ミドルウェアを非同期で定義できるようになりました。非同期ミドルウェアは ASGI サーバのイベントループ上で直接実行され、スレッドプールへのオフロードが発生しません。 ```python class AsyncTimingMiddleware: # async_capable = True はクラス属性で宣言可能だが、 # __acall__ を定義すれば Django は自動的に非同期と判定する def __init__(self, get_response): self.get_response = get_response async def __call__(self, request): import time start = time.time() response = await self.get_response(request) elapsed = (time.time() - start) * 1000 response["X-Process-Time"] = f"{elapsed:.1f}ms" return response ``` 同期版との違いは `def __call__` が `async def __call__` に変わり、`self.get_response(request)` が `await self.get_response(request)` になった点だけです。Django は `asyncio.iscoroutinefunction(middleware.__call__)` で非同期かどうかを判定し、非同期であればラップなしで直接チェーンに組み込みます。 非同期ミドルウェアの利点は、前後処理で `await` を使えることです。 ```python import httpx import time class AsyncAuthMiddleware: def __init__(self, get_response): self.get_response = get_response async def __call__(self, request): token = request.headers.get("Authorization", "") if token.startswith("Bearer "): # 外部認証サービスへの非同期呼び出し async with httpx.AsyncClient() as client: result = await client.post( "https://auth.example.com/verify", json={"token": token[7:]}, timeout=5, ) if result.status_code == 200: request.verified_user = result.json() else: request.verified_user = None else: request.verified_user = None response = await self.get_response(request) return response ``` 同じ処理を同期ミドルウェアで実装すると、`requests.post()` でブロッキングが発生します。ASGI サーバ上では `sync_to_async` でスレッドプールに逃がされますが、非同期ミドルウェアであればイベントループ上で `await` し、待機中に他のリクエストを処理できます。 同期・非同期の両方に対応するミドルウェアを書くことも可能です。Django は `sync_capable` と `async_capable` のクラス属性を検査し、実行環境に応じた呼び出し方を選択します。 ```python from asgiref.sync import iscoroutinefunction, markcoroutinefunction, sync_to_async class DualModeTimingMiddleware: sync_capable = True async_capable = True def __init__(self, get_response): self.get_response = get_response if iscoroutinefunction(self.get_response): markcoroutinefunction(self) def __call__(self, request): if iscoroutinefunction(self): return self.__acall__(request) import time start = time.time() response = self.get_response(request) elapsed = (time.time() - start) * 1000 response["X-Process-Time"] = f"{elapsed:.1f}ms" return response async def __acall__(self, request): import time start = time.time() response = await self.get_response(request) elapsed = (time.time() - start) * 1000 response["X-Process-Time"] = f"{elapsed:.1f}ms" return response ``` `markcoroutinefunction(self)` は `asyncio.iscoroutinefunction(self.__call__)` が `True` を返すようにマーキングします。Django はこのフラグを見てチェーン構築時に同期ラップをスキップします。Django 自身の組み込みミドルウェア（`SecurityMiddleware`、`SessionMiddleware`、`AuthenticationMiddleware` など）もこのデュアルモードパターンに順次対応しつつあります。 ### 混在時の考え方 実務のプロジェクトでは、同期ミドルウェアと非同期ミドルウェアが混在する状況が一般的です。 ```python MIDDLEWARE = [ 'django.middleware.security.SecurityMiddleware', # デュアルモード 'myapp.middleware.AsyncLoggingMiddleware', # 非同期 'django.contrib.sessions.middleware.SessionMiddleware', # デュアルモード 'django.middleware.common.CommonMiddleware', # デュアルモード 'django.middleware.csrf.CsrfViewMiddleware', # デュアルモード 'django.contrib.auth.middleware.AuthenticationMiddleware', # デュアルモード 'myapp.middleware.LegacySyncMiddleware', # 同期のみ 'myapp.middleware.AsyncAuthMiddleware', # 非同期 ] ``` ASGI サーバ上でこのチェーンが構築されるとき、Django は各ミドルウェアの同期・非同期を個別に判定し、必要に応じて `sync_to_async` でラップします。チェーン全体の実行の流れを追うと次のようになります。 ```{mermaid} flowchart TD EL["ASGI イベントループ"] EL --> SM1["SecurityMiddleware
async 版 → 直接実行"] SM1 --> AM1["AsyncLoggingMiddleware
async → 直接実行"] AM1 --> SM2["SessionMiddleware
async 版 → 直接実行"] SM2 --> LS["LegacySyncMiddleware
同期 → sync_to_async でラップ
スレッドプール実行"] LS --> AM2["AsyncAuthMiddleware
async → 直接実行"] AM2 --> VW["ビュー関数"] ``` ``` ASGI イベントループ │ ▼ SecurityMiddleware (async 版が選択される) │ → イベントループ上で実行 ▼ AsyncLoggingMiddleware (async) │ → イベントループ上で実行 ▼ SessionMiddleware (async 版が選択される) │ → イベントループ上で実行、DB アクセスは内部で sync_to_async ▼ CommonMiddleware (async 版が選択される) │ → イベントループ上で実行 ▼ CsrfViewMiddleware (async 版が選択される) │ → イベントループ上で実行 ▼ AuthenticationMiddleware (async 版が選択される) │ → イベントループ上で実行、ユーザー取得は内部で sync_to_async ▼ LegacySyncMiddleware (sync → sync_to_async でラップ) │ → スレッドプールで実行 ← ここでコンテキストスイッチ ▼ AsyncAuthMiddleware (async) │ → イベントループ上で実行 ▼ ビュー関数 ``` `LegacySyncMiddleware` の位置でイベントループからスレッドプールへのコンテキストスイッチが発生します。そしてその次の `AsyncAuthMiddleware` でスレッドプールからイベントループへ戻ります。この往復にはオーバーヘッドがあります。 リクエストごとに同期・非同期の境界を何度も越えると、そのたびにスレッドの取得・解放とコンテキストの保存・復元が発生します。ミドルウェアが8つあり、そのうち3つが同期だとすると、最大で6回の境界越え（同期→非同期→同期→非同期→…）が起こりえます。 この問題を軽減するための現実的なアプローチを紹介します。 - **同期ミドルウェアをチェーンの一方の端にまとめる**: 境界越えの回数を減らせます - **すべてのミドルウェアを非同期対応にする**: デュアルモード含め非同期対応であれば、境界越えは発生しません。Django の組み込みミドルウェアはバージョンアップのたびにデュアルモード対応が進んでいるため、残る問題はサードパーティ製の同期ミドルウェアです - **同期ミドルウェアを非同期版に書き換える**: サードパーティの同期ミドルウェアが処理の大部分を占めている場合に有効です。ただし内部で ORM にアクセスする場合は `sync_to_async` が必要になるため、完全な非同期化は ORM の非同期対応状況に依存します パフォーマンスへの影響を測定するには、{numref}`Django の request 処理と async`（{ref}`Django の request 処理と async`）で示したように各ミドルウェアの前後で時間を計測します。 ```python class MiddlewareProfilingMiddleware: def __init__(self, get_response): self.get_response = get_response async def __call__(self, request): import time start = time.time() response = await self.get_response(request) elapsed = (time.time() - start) * 1000 # ミドルウェアチェーン以下の合計処理時間 print(f"[Profile] {request.path} inner chain: {elapsed:.1f}ms") return response ``` このプロファイリングミドルウェアをチェーンの異なる位置に挿入し、同期ミドルウェアを挟む前後で処理時間がどの程度変わるかを比較することで、境界越えのオーバーヘッドを定量的に把握できます。 Django の ASGI ミドルウェア対応は「既存の同期コードを壊さずに動かす」ことを最優先にしつつ、「新しいコードは非同期で書ける」余地を提供する段階的移行戦略です。{numref}`ASGI ミドルウェアを書く`（{ref}`ASGI ミドルウェアを書く`）で手書きした ASGI ミドルウェアが `async def __call__(self, scope, receive, send)` で `send` をラップしていたのに対し、Django のミドルウェアは `request` / `response` オブジェクトを介した高レベルの抽象化を維持しています。これにより開発者は `scope` や `receive` / `send` を直接扱う必要がなく、同期・非同期の切り替えも `def __call__` を `async def __call__` に変えるだけで済みます。 次節では Django Channels による WebSocket 対応を見ていきます。 (Django で ASGI を使う場面)= ## Django で ASGI を使う場面 ### async view {numref}`Django の request 処理と async`（{ref}`Django の request 処理と async`）で非同期ビューの仕組みを追いましたが、ここでは「どのようなケースで `async def` ビューを選択すべきか」という実務上の判断基準を整理します。 非同期ビューが効果を発揮するのは、**ビュー関数内で複数の独立した I/O 待ちが発生するケース**です。 ```python import httpx import asyncio from django.http import JsonResponse async def dashboard(request): async with httpx.AsyncClient() as client: # 3つの外部 API を並行して呼び出す user_stats, sales_data, notifications = await asyncio.gather( client.get("https://api.example.com/user-stats", timeout=5), client.get("https://api.example.com/sales", timeout=5), client.get("https://api.example.com/notifications", timeout=5), ) return JsonResponse({ "user_stats": user_stats.json(), "sales": sales_data.json(), "notifications": notifications.json(), }) ``` 同期ビューと非同期ビューの処理時間を比べると以下のようになります。 | 方式 | 処理時間の目安 | 理由 | |------|-------------|------| | 同期ビュー | 600ms（各 200ms × 3） | API 呼び出しが逐次実行される | | 非同期ビュー（`asyncio.gather`） | ≈200ms（最も遅い1件） | 3つが並行実行される | これは同期ビューをスレッドプールで実行しても得られない利点です。スレッドプールはビュー関数全体を1つのスレッドで実行するため、関数内部の並行化は行われません。 ```{mermaid} sequenceDiagram participant V as async view participant A1 as 外部 API 1 participant A2 as 外部 API 2 participant A3 as 外部 API 3 V->>A1: GET /user-stats V->>A2: GET /sales V->>A3: GET /notifications Note over V,A3: asyncio.gather で3つを並行送信 A1-->>V: 200ms A2-->>V: 180ms A3-->>V: 200ms Note over V: 合計 ≈ 200ms (最大値のみ待つ) ``` 一方で、単一の ORM クエリを実行して結果を返すだけのビューでは、非同期にする利点はほとんどありません。 ```{tip} `def` ビューと `async def` ビューを使い分ける目安: 外部 I/O が1つだけなら `def`、複数の外部 I/O を並行して待つ場合は `async def` を検討してください。 ``` ```python # このビューを async def にする必要性は低い def user_detail(request, user_id): user = User.objects.select_related("profile").get(id=user_id) return JsonResponse({"id": user.id, "name": user.name}) ``` ORM クエリは1回の I/O 待ちで完了し、並行化の余地がありません。{numref}`Django の request 処理と async`（{ref}`Django の request 処理と async`）で述べた通り、Django の非同期 ORM は内部的に `sync_to_async` でラップされているため、`await User.objects.aget(id=user_id)` としてもデータベースドライバレベルでは同期のままです。このケースでは `def` のまま残し、ASGI 上では Django が自動的にスレッドプールで実行する方がシンプルです。 ### 長時間待ち I/O 非同期ビューが特に有効なのは、レスポンスの生成に長い I/O 待ちを伴うケースです。 Server-Sent Events（SSE）によるストリーミングはその典型的な例です。 ```python import asyncio from django.http import StreamingHttpResponse async def sse_notifications(request): async def event_stream(): while True: # 新しい通知を非同期でポーリング notifications = await get_new_notifications(request.user.id) if notifications: for n in notifications: yield f"data: {json.dumps(n)}\n\n" await asyncio.sleep(2) response = StreamingHttpResponse( event_stream(), content_type="text/event-stream", ) response["Cache-Control"] = "no-cache" response["X-Accel-Buffering"] = "no" return response ``` 同期ビューと非同期ビューで SSE エンドポイントを実装した場合の違いを示します。 - **同期ビュー**: `time.sleep(2)` がワーカースレッドを占有し続けます。Gunicorn のワーカー数が4であれば、4つの SSE 接続でサーバの処理能力が飽和します - **非同期ビュー**: `await asyncio.sleep(2)` がイベントループを解放するため、数百の SSE 接続を単一のワーカープロセスで処理できます LLM のトークンストリーミングも同様のパターンです。 ```python async def chat_stream(request): body = json.loads(request.body) prompt = body["prompt"] async def generate(): async with httpx.AsyncClient() as client: async with client.stream( "POST", "https://api.openai.com/v1/chat/completions", json={"model": "gpt-4", "messages": [{"role": "user", "content": prompt}], "stream": True}, headers={"Authorization": f"Bearer {settings.OPENAI_API_KEY}"}, timeout=60, ) as response: async for line in response.aiter_lines(): if line.startswith("data: ") and line != "data: [DONE]": yield f"{line}\n\n" return StreamingHttpResponse(generate(), content_type="text/event-stream") ``` 外部 API からのストリーミングレスポンスをそのままクライアントに中継する処理は、非同期 I/O の恩恵を直接受けます。`async for line in response.aiter_lines()` がチャンクを待っている間、イベントループは他のリクエストを処理し続けます。 バッチ処理的な外部 API 呼び出しも、非同期ビューが効果的な場面です。 ```python async def enrich_users(request): user_ids = json.loads(request.body)["user_ids"] async with httpx.AsyncClient() as client: tasks = [ client.get(f"https://api.example.com/profile/{uid}", timeout=10) for uid in user_ids ] responses = await asyncio.gather(*tasks, return_exceptions=True) results = [] for uid, resp in zip(user_ids, responses): if isinstance(resp, Exception): results.append({"user_id": uid, "error": str(resp)}) else: results.append({"user_id": uid, "profile": resp.json()}) return JsonResponse({"results": results}) ``` 20件のユーザープロフィールを外部 API から取得する場合、同期では逐次実行で数秒かかる処理が、非同期の `asyncio.gather` で並行実行すれば最も遅い1件の応答時間程度で完了します。 ### WebSocket は素の Django だけでは足りないこと {numref}`Django は ASGI にどう対応しているか`（{ref}`Django は ASGI にどう対応しているか`）で見た通り、Django の `ASGIHandler` は `scope["type"] == "http"` と `scope["type"] == "lifespan"` を処理しますが、`scope["type"] == "websocket"` には対応していません。 ```{warning} 素の Django（Channels なし）で WebSocket 接続を受け付けると、`ASGIHandler` が `ValueError: Unknown scope type: websocket` を送出します。WebSocket が必要な場合は Django Channels を導入してください。 ``` ```python # Django の ASGIHandler（概念コード） async def __call__(self, scope, receive, send): if scope["type"] == "http": await self.handle(scope, receive, send) elif scope["type"] == "lifespan": # lifespan 処理 ... else: raise ValueError(f"Unknown scope type: {scope['type']}") # ↑ websocket が来るとここで例外 ``` {numref}`WebSocket と lifespan`（{ref}`WebSocket と lifespan`）と{numref}`WebSocket の最小実装`（{ref}`WebSocket の最小実装`）で見たように、WebSocket は `websocket.connect` → `websocket.accept` → メッセージの `receive` / `send` ループ → `websocket.disconnect` というライフサイクルを持ちます。HTTP とはまったく異なるプロトコルであり、Django のミドルウェアチェーン、URL 解決、ビュー関数という HTTP 前提の処理パイプラインにそのまま組み込むことはできません。 Django が WebSocket を標準サポートしない理由は、Django の設計がリクエスト・レスポンスモデルに深く根差しているためです。 - `HttpRequest` / `HttpResponse` という抽象化 - ミドルウェアの `__call__(self, request)` → `response` という構造 - URL 解決からビュー実行までの一方向フロー これらはすべて「1つのリクエストに対して1つのレスポンスを返す」HTTP の前提に基づいています。WebSocket の双方向・長時間接続モデルはこの前提を根本的に覆すため、Django のコアに組み込むには大幅な設計変更が必要です。 ### Channels の位置づけを軽く紹介 Django Channels は、Django に WebSocket と他の非 HTTP プロトコルのサポートを追加するための公式拡張パッケージです。Django プロジェクトの一部として開発されていますが、Django 本体には含まれていません。 ```bash pip install channels ``` Channels の核心的なアイデアは、Django の `ASGIHandler` を別の ASGI アプリケーション（`ProtocolTypeRouter`）でラップし、`scope["type"]` に応じて処理を振り分けることです。 ```{note} `ProtocolTypeRouter` は{numref}`ルーティングを自作してみる`（{ref}`ルーティングを自作してみる`）で自作したルーターの `scope["type"]` 分岐と同じ発想です。HTTP リクエストは従来通り Django の `ASGIHandler` が処理し、WebSocket 接続は Channels の `URLRouter` が処理します。 ``` ```python # myproject/asgi.py（Channels 使用時） import os from django.core.asgi import get_asgi_application from channels.routing import ProtocolTypeRouter, URLRouter from channels.auth import AuthMiddlewareStack from myapp.routing import websocket_urlpatterns os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings') django_asgi_app = get_asgi_application() application = ProtocolTypeRouter({ "http": django_asgi_app, "websocket": AuthMiddlewareStack( URLRouter(websocket_urlpatterns) ), }) ``` ```python # myapp/routing.py from django.urls import path from myapp.consumers import ChatConsumer websocket_urlpatterns = [ path("ws/chat//", ChatConsumer.as_asgi()), ] ``` ```python # myapp/consumers.py from channels.generic.websocket import AsyncJsonWebsocketConsumer class ChatConsumer(AsyncJsonWebsocketConsumer): async def connect(self): self.room_name = self.scope["url_route"]["kwargs"]["room_name"] self.room_group = f"chat_{self.room_name}" await self.channel_layer.group_add(self.room_group, self.channel_name) await self.accept() async def disconnect(self, close_code): await self.channel_layer.group_discard(self.room_group, self.channel_name) async def receive_json(self, content): await self.channel_layer.group_send( self.room_group, {"type": "chat.message", "message": content["message"], "sender": self.scope["user"].username}, ) async def chat_message(self, event): await self.send_json({ "message": event["message"], "sender": event["sender"], }) ``` Channels の Consumer は{numref}`WebSocket の最小実装`（{ref}`WebSocket の最小実装`）で手書きした WebSocket ハンドラを高レベルに抽象化したものです。 | Consumer メソッド | 対応する生のイベント | |----------------|----------------| | `connect` | `websocket.connect` | | `disconnect` | `websocket.disconnect` | | `receive_json` | `websocket.receive` | | `send_json` | `websocket.send` | `self.scope` は ASGI の `scope` そのものであり、`self.scope["user"]` は `AuthMiddlewareStack` がセッション情報から解決した Django のユーザーオブジェクトです。 Channels が提供する Channel Layer（Redis バックエンドが一般的）は、複数のワーカープロセスやサーバ間でメッセージをブロードキャストするための Pub/Sub 機構です。{numref}`WebSocket の最小実装`（{ref}`WebSocket の最小実装`）のチャットルーム実装では `connected` 集合にインメモリで `send` callable を保持しましたが、これは単一プロセスでしか動作しません。Channel Layer はプロセスの壁を越えてメッセージを配信します。 本書では Channels の詳細な実装には踏み込みませんが、重要なのは Channels が Django のコアを変更するのではなく、ASGI のプロトコルルーティングを活用して Django の横に WebSocket 処理を並置する設計であるという点です。Django 本体は依然として HTTP 処理に集中し、WebSocket は Channels が受け持つという責務分離が、ASGI の `scope["type"]` 分岐によって実現されています。 次節では Django の ASGI 対応に関するトラブルシューティングの観点を整理します。 (Django ASGI の限界と誤解しやすい点)= ## Django ASGI の限界と誤解しやすい点 ### 「ASGI 対応」と「全部 async で速い」は違う Django を ASGI サーバ上で動かした瞬間にすべてが非同期になり高速化されるという期待は、最もよくある誤解です。 ```{caution} `uvicorn myproject.asgi:application` で起動しただけでは、既存の同期ビューと同期ミドルウェアの実行方式が変わるだけです。「ASGI にすれば速くなる」という期待は誤りです。 ``` {numref}`Django の request 処理と async`（{ref}`Django の request 処理と async`）で見た通り、同期ビューは `sync_to_async` でスレッドプールにオフロードされます。これは WSGI サーバ（Gunicorn）がワーカースレッドでビューを実行するのと本質的に同じことです。スレッドプールのスレッド数には上限があり、同時に処理できるリクエスト数はその上限に制約されます。 ``` WSGI (Gunicorn --workers 4 --threads 4): → 最大 16 リクエストを同時処理（4 プロセス × 4 スレッド） ASGI (Uvicorn) + 全ビューが同期 def: → スレッドプールのデフォルトサイズ（40）が同時処理の上限 → イベントループ自体は空いているが、実際の処理はスレッドプール内 ``` むしろ、同期コードだけのプロジェクトを ASGI に移行すると、`sync_to_async` のラップに伴うオーバーヘッド（スレッド取得・解放、コンテキスト保存・復元）が追加されるため、単純なベンチマークでは WSGI よりわずかに遅くなる場合があります。 ASGI の利点が発揮されるのは、{numref}`Django で ASGI を使う場面`（{ref}`Django で ASGI を使う場面`）で述べた「複数の外部 API を並行呼び出しする」「SSE やストリーミングで長時間接続を保持する」「WebSocket を使う」といった、非同期 I/O が本質的に必要な場面に限られます。 ASGI へ移行すべきかどうかの判断基準を整理します。 | 状況 | 推奨 | |------|------| | 非同期 I/O が必要なエンドポイントがある | ASGI へ移行 | | WebSocket や SSE の要件がある | ASGI へ移行 | | 将来的に非同期化を段階的に進める計画がある | ASGI へ移行 | | ORM 中心の CRUD アプリケーション | WSGI のまま運用 | | 管理画面が主体のシステム | WSGI のまま運用 | | サードパーティパッケージが同期前提のプロジェクト | WSGI のまま運用 | ```{mermaid} flowchart LR Q{"プロジェクトの特性"} Q -->|外部 API 並行呼び出し
SSE / WebSocket が必要| ASGI["ASGI へ移行
非同期の恩恵を享受"] Q -->|ORM 中心の CRUD
管理画面が主体
サードパーティが同期前提| WSGI["WSGI のまま運用
安定性・簡潔性を優先"] ``` ### ORM と async の注意点 Django ORM の非同期対応は進行中ですが、現時点ではいくつかの重要な制約があります。 {numref}`Django の request 処理と async`（{ref}`Django の request 処理と async`）で述べた通り、非同期 ORM メソッド（`aget`, `acreate`, `asave` など）の多くは内部的に `sync_to_async` で同期版をラップしています。データベースドライバ（`psycopg2`, `mysqlclient` など）自体が同期のため、`await User.objects.aget(id=1)` はスレッドプール内で `User.objects.get(id=1)` を実行し、その完了を `await` で待つ構造です。 ```{important} 真の非同期データベースアクセス（ドライバレベルでの non-blocking I/O）は実現されていません。非同期 ORM に切り替えたからといってデータベースアクセスが高速化されるわけではなく、利点はイベントループがブロックされないことにあります。 ``` ```python # これらは見た目は非同期だが、内部的にはスレッドプール経由の同期実行 user = await User.objects.aget(id=1) # sync_to_async(User.objects.get)(id=1) count = await User.objects.filter(active=True).acount() # sync_to_async(qs.count)() await user.asave() # sync_to_async(user.save)() ``` 次に、非同期ビュー内で同期 ORM メソッドを呼ぶと `SynchronousOnlyOperation` 例外が発生します。これは意図的な安全機構ですが、実務では思わぬ場所でこの例外に遭遇します。 ```text async def user_detail(request, user_id): user = await User.objects.aget(id=user_id) # OK # ここで問題が起きやすい profile = user.profile # ForeignKey の遅延読み込み → SynchronousOnlyOperation! ``` `user.profile` は Django ORM の遅延読み込み（lazy loading）を発動し、内部的に同期の SQL クエリを実行します。非同期コンテキストではこれが検出されて例外になります。 対策は2つあります。 1. `select_related` で JOIN を事前に指定する 2. 関連オブジェクトのアクセスを `sync_to_async` でラップする ```text async def user_detail(request, user_id): # select_related で事前にJOIN → 遅延読み込みが発生しない user = await User.objects.select_related("profile").aget(id=user_id) profile = user.profile # 既に読み込み済みなので同期アクセスでも OK # または sync_to_async でラップ @sync_to_async def get_user_with_relations(uid): user = User.objects.get(id=uid) return {"id": user.id, "name": user.name, "profile": user.profile.bio} data = await get_user_with_relations(user_id) return JsonResponse(data) ``` さらに、`QuerySet` の構築自体は同期でも非同期でも安全ですが、評価のタイミングには注意が必要です。 ```python async def user_list(request): qs = User.objects.filter(active=True).order_by("name") # QuerySet の構築（SQL 未実行） # 評価方法の選択 users = await sync_to_async(list)(qs) # 方法1: list() を sync_to_async # または users = [user async for user in qs] # 方法2: async for（Django 4.1+） # または users = await sync_to_async(lambda: list(qs.values("id", "name")))() # 方法3 return JsonResponse({"users": users}) ``` `async for user in qs` は Django 4.1 以降で使える構文で、QuerySet が `__aiter__` を実装しています。内部的にはチャンク単位で `sync_to_async` を使ってレコードを取得します。 ### middleware やサードパーティパッケージの対応状況 Django エコシステムの大部分は同期を前提に構築されています。ASGI に移行する際、自分のコード以外の部分が問題の原因になることがあります。 Django 自身の組み込みミドルウェアはバージョンアップのたびにデュアルモード（同期・非同期両対応）化が進んでいます。{numref}`middleware は async にどう対応するか`（{ref}`middleware は async にどう対応するか`）で見た通り、`SecurityMiddleware`、`SessionMiddleware`、`CommonMiddleware`、`CsrfViewMiddleware`、`AuthenticationMiddleware` などは非同期対応が進んでいます。しかし、これらのミドルウェアが内部で ORM にアクセスする際（セッションの読み書き、ユーザーの取得など）は `sync_to_async` を経由するため、完全な非同期実行にはなっていません。 サードパーティパッケージの対応状況はまちまちです。 ```python MIDDLEWARE = [ 'django.middleware.security.SecurityMiddleware', # デュアルモード ✓ 'corsheaders.middleware.CorsMiddleware', # 同期のみの場合あり △ 'django.contrib.sessions.middleware.SessionMiddleware', # デュアルモード ✓ 'django.middleware.common.CommonMiddleware', # デュアルモード ✓ 'django.middleware.csrf.CsrfViewMiddleware', # デュアルモード ✓ 'django.contrib.auth.middleware.AuthenticationMiddleware', # デュアルモード ✓ 'debug_toolbar.middleware.DebugToolbarMiddleware', # 同期前提 ✗ 'myapp.middleware.CustomMiddleware', # 自分次第 ] ``` 同期のみのサードパーティミドルウェアは、{numref}`middleware は async にどう対応するか`（{ref}`middleware は async にどう対応するか`）で述べた通り `sync_to_async` で自動ラップされるため動作はします。しかし、チェーン内の同期・非同期の境界越えが増え、パフォーマンスに影響する可能性があります。 Django REST Framework（DRF）の非同期対応も段階的に進行しています。DRF のビューやシリアライザは歴史的に同期で設計されており、非同期ビューとの組み合わせには制約があります。DRF を使用するプロジェクトでは、API エンドポイントの大部分が DRF のクラスベースビューで実装されているため、その非同期対応状況がプロジェクト全体の ASGI 移行可否に大きく影響します。 `django-celery`、`django-redis`、`django-storages` などの I/O を伴うパッケージも、同期 API のみを提供しているものが多くあります。非同期ビュー内からこれらを使う場合は `sync_to_async` でラップする必要があり、完全な非同期化の障壁になります。 実務上のアプローチを整理します。 - プロジェクト内で使用しているすべてのサードパーティパッケージの非同期対応状況を事前に調査する - 対応していないパッケージが多い場合は WSGI のまま運用することを検討する - 非同期が必要なエンドポイントだけを別サービスとして FastAPI で実装し、Django とは API ゲートウェイやリバースプロキシで結合するという構成も現実的な選択肢です ```{admonition} Django ASGI の現在地 :class: note Django の ASGI 対応は「すべてを一度に非同期化する」ことを目指していません。Django のコア開発チームは DEP（Django Enhancement Proposal）を通じて段階的な非同期化ロードマップを示しており、ORM の完全非同期化、ミドルウェアの完全非同期化、テンプレートレンダリングの非同期化などが将来的に計画されています。現時点での Django ASGI は「完成した非同期フレームワーク」ではなく「同期から非同期へ移行する途中の状態」であり、その制約を理解した上で使うことが重要です。 ``` 次節では、本章のまとめとして Django の ASGI 対応全体を振り返り、トラブルシューティングの観点で整理します。 (ch10-トラブルシューティングの観点)= ## トラブルシューティングの観点 ### SynchronousOnlyOperation Django の ASGI 環境で最も頻繁に遭遇する例外が `django.core.exceptions.SynchronousOnlyOperation` です。このエラーは「非同期コンテキスト内で同期的なデータベースアクセスまたはその他のブロッキング操作が検出された」ことを意味します。 ``` django.core.exceptions.SynchronousOnlyOperation: You cannot call this from an async context - use a thread or sync_to_async. ``` この例外が送出される典型的な場面を整理します。 **最も多いのは、非同期ビュー内での同期 ORM 呼び出しです。** ```python async def user_detail(request, user_id): user = User.objects.get(id=user_id) # SynchronousOnlyOperation return JsonResponse({"id": user.id}) ``` 対処は `aget` を使うか、`sync_to_async` でラップすることです。ただし、`aget` に書き換えただけでは解決しないケースがあります。{numref}`Django ASGI の限界と誤解しやすい点`（{ref}`Django ASGI の限界と誤解しやすい点`）で述べた遅延読み込みの問題です。 ```python async def user_detail(request, user_id): user = await User.objects.aget(id=user_id) # ← ここは OK return JsonResponse({ "id": user.id, "name": user.name, "department": user.department.name, # ← SynchronousOnlyOperation! }) ``` `user.department` は ForeignKey のアクセスであり、`select_related` を指定していなければ遅延読み込みが発動します。遅延読み込みは内部的に同期の SQL クエリを実行するため、非同期コンテキストで検出されて例外になります。エラーメッセージには `user.department` がどこで呼ばれたかのスタックトレースが含まれますが、ORM の内部スタックが深いため、一見するとどの行が原因か分かりにくいことがあります。 対処のパターンは3つあります。 ```{mermaid} flowchart TD SOO["SynchronousOnlyOperation 発生"] SOO --> LC["スタックトレースで
発生箇所を特定"] LC --> FK{"ForeignKey
遅延読み込み?"} FK -->|Yes| SR["select_related で
事前 JOIN 読み込み"] FK -->|No| DP{"対象が ORM
操作?"} DP -->|Yes| S2A["sync_to_async で
まとめてラップ"] DP -->|No| AOP["aget / acreate など
非同期版 ORM に変更"] ``` ```python # パターン1: select_related で事前読み込み async def user_detail(request, user_id): user = await User.objects.select_related("department").aget(id=user_id) return JsonResponse({"id": user.id, "department": user.department.name}) # パターン2: 関連データ含めて sync_to_async でまとめて取得 @sync_to_async def get_user_data(user_id): user = User.objects.select_related("department").get(id=user_id) return {"id": user.id, "name": user.name, "department": user.department.name} async def user_detail(request, user_id): data = await get_user_data(user_id) return JsonResponse(data) # パターン3: values / values_list で必要なフィールドだけ取得 async def user_detail(request, user_id): data = await User.objects.filter(id=user_id).select_related( "department" ).values("id", "name", "department__name").afirst() return JsonResponse(data) ``` ```{tip} パターン2が最も安全です。同期コンテキスト内で ORM の操作を完結させ、Python の基本型（辞書やリスト）として返すことで、非同期コンテキストに ORM オブジェクトが持ち込まれるのを防ぎます。ORM オブジェクトを非同期コンテキストに持ち込むと、テンプレートや後続の処理で遅延読み込みが意図せず発動するリスクがあります。 ``` `SynchronousOnlyOperation` は ORM だけでなく、キャッシュバックエンド（`cache.get()`）やファイル操作など、Django が同期専用と判定した操作でも発生します。Django は内部的に `asyncio.get_running_loop()` でイベントループの有無を検査し、イベントループ内であれば同期操作を拒否する仕組みを持っています。 ### event loop まわりの誤解 ASGI 環境でのイベントループに関する誤解は、デバッグを困難にする原因になります。 最も多い誤解は「`asyncio.run()` で非同期コードを実行すればよい」というものです。 ```{danger} ASGI サーバ上の同期ビュー内で `asyncio.run()` を使ってはいけません。Python の `asyncio` は同一スレッドに複数のイベントループを許容しないため、`RuntimeError: This event loop is already running` が発生します。代わりに `async_to_sync` を使ってください。 ``` ```python import asyncio async def fetch_data(): async with httpx.AsyncClient() as client: return await client.get("https://api.example.com/data") # 同期ビュー内で asyncio.run() を使う def bad_view(request): result = asyncio.run(fetch_data()) # RuntimeError の可能性 return JsonResponse(result.json()) ``` ASGI サーバ上では既にイベントループが動作しています。WSGI サーバ上では動作する場合がありますが、それはイベントループが存在しないためです。環境によって動いたり動かなかったりするコードは、本番で障害を引き起こします。 正しい対処は `async_to_sync` を使うことです。 ```python from asgiref.sync import async_to_sync def correct_view(request): result = async_to_sync(fetch_data)() return JsonResponse(result.json()) ``` `async_to_sync` は現在のスレッドにイベントループが存在する場合としない場合の両方を正しく処理します。{numref}`sync / async 境界の橋渡し`（{ref}`sync / async 境界の橋渡し`）で述べた通り、イベントループが存在する場合は別スレッドで実行し、存在しない場合は新しいイベントループを作成します。 もう一つの誤解は、`asyncio.get_event_loop()` を使ってタスクを手動でスケジュールしようとするケースです。 ```python async def fire_and_forget_bad(request): loop = asyncio.get_event_loop() loop.create_task(send_notification(request.user.id)) # リクエスト完了後にタスクが消える可能性 return JsonResponse({"status": "accepted"}) ``` `create_task` でスケジュールされたタスクは、リクエストのライフサイクルとは無関係にイベントループ上で実行されます。しかし、タスクの完了を待つ仕組みがないため、タスク内で例外が発生してもログに出るだけで、サーバのシャットダウン時にタスクが中断される可能性があります。バックグラウンド処理が必要な場合は Celery などのタスクキューを使うか、タスクの完了を適切に管理する仕組みを導入すべきです。 `DJANGO_ALLOW_ASYNC_UNSAFE` 環境変数に関する誤解もあります。 ```python # 開発時のシェルで見かける設定 os.environ["DJANGO_ALLOW_ASYNC_UNSAFE"] = "true" ``` この環境変数は `SynchronousOnlyOperation` のチェックを無効化します。Django シェル（`manage.py shell`）や Jupyter Notebook のように、イベントループが自動的に作成される環境で同期 ORM を使いたい場合の回避策です。 ```{danger} `DJANGO_ALLOW_ASYNC_UNSAFE=true` を本番環境に設定してはいけません。非同期ビュー内の同期 ORM 呼び出しが例外にならず、イベントループのブロッキングが黙って発生します。{numref}`ch07-トラブルシューティングの観点`（{ref}`ch07-トラブルシューティングの観点`）や {numref}`ch09-トラブルシューティングの観点`（{ref}`ch09-トラブルシューティングの観点`）で述べた「async 内で blocking」の問題がフレームワークの安全弁なしに起きます。 ``` ### async view に blocking 処理を入れる問題 `SynchronousOnlyOperation` は Django が検出できるブロッキングのみを対象としています。Django の管理下にない同期ブロッキング呼び出しは検出されず、黙ってイベントループを停止させます。 ```python import requests # Django は requests.get のブロッキングを検出しない import time async def external_api(request): # 以下の行でイベントループが最大5秒間ブロックされる response = requests.get("https://api.example.com/slow", timeout=5) return JsonResponse(response.json()) async def slow_view(request): time.sleep(3) # イベントループが3秒間ブロックされる return JsonResponse({"status": "done"}) ``` Django の `SynchronousOnlyOperation` は ORM、キャッシュ、シグナルなど Django 内部の操作のみを検出します。`requests`、`time.sleep`、`open()` によるファイル読み書き、`subprocess.run()` などの標準ライブラリや外部ライブラリのブロッキング呼び出しは検出されません。 この問題の厄介さは、単体テストや低負荷環境ではまったく問題が表面化しないことです。 - 1リクエストずつ処理する限り、ブロッキングの影響は自分自身のレスポンスタイムに限定されます - 同時リクエスト数が増えて初めて、他のリクエストのレスポンスタイムが連鎖的に悪化します 検出方法として最も有効なのは、{numref}`sync / async 境界の橋渡し`（{ref}`sync / async 境界の橋渡し`）でも触れた `PYTHONASYNCIODEBUG=1` 環境変数です。 ```bash PYTHONASYNCIODEBUG=1 uvicorn myproject.asgi:application --port 8000 ``` この設定により、イベントループ上で100ms以上ブロックする処理が警告として出力されます。開発環境で有効にしておけば、ブロッキング呼び出しを早期に発見できます。 もう一つの検出方法は、ミドルウェアでリクエスト処理時間を計測し、異常に長い処理を記録することです。 ```python class AsyncBlockingDetectorMiddleware: def __init__(self, get_response): self.get_response = get_response async def __call__(self, request): import time start = time.monotonic() response = await self.get_response(request) elapsed = time.monotonic() - start if elapsed > 1.0: import logging logger = logging.getLogger("blocking_detector") logger.warning( f"Slow request: {request.method} {request.path} took {elapsed:.2f}s" ) return response ``` ブロッキングを発見した場合の対処は、{numref}`ch09-トラブルシューティングの観点`（{ref}`ch09-トラブルシューティングの観点`）で述べた3つのパターンと同じです。 1. **非同期ライブラリへの置き換え**（最善）: `requests` → `httpx`、`time.sleep` → `asyncio.sleep`、`open` → `aiofiles` 2. **スレッドプールへの逃避**: 置き換えが困難な場合は `sync_to_async(thread_sensitive=False)` でスレッドプールに逃がします 3. **`def` ビューに変更**: そもそも `async def` にする必要がないビューであれば `def` に変更し、Django の自動スレッドプール実行に委ねます ```{important} `async def` ビューの中で呼ぶすべての I/O 関数に `await` が付いているかを確認することが判断の原則です。`await` なしの I/O 呼び出しが1つでもあれば、それはブロッキングです。Django が検出して例外にしてくれるのは ORM 関連のみであり、それ以外は開発者の責任で管理する必要があります。 ``` ブロッキング呼び出しの検出可否を整理します。 | 操作 | 検出可否 | |------|---------| | Django ORM（`.get`, `.filter`, `.save`） | `SynchronousOnlyOperation` で検出される | | Django キャッシュ（`cache.get`, `cache.set`） | `SynchronousOnlyOperation` で検出される | | `requests.get` / `urllib` | 検出されない | | `time.sleep` | 検出されない | | `open()` / ファイル読み書き | 検出されない | | `subprocess.run` | 検出されない | | 同期データベースドライバ直接呼び出し | 検出されない | 本章を通じて、Django の ASGI 対応は「同期コードベースとの互換性を維持しながら非同期の利点を段階的に取り入れる」という設計方針であることを見てきました。`ASGIHandler` による入口の変換、`sync_to_async` / `async_to_sync` によるブリッジ、ミドルウェアの自動ラップと `SynchronousOnlyOperation` による安全弁、そしてそれでも検出できないブロッキングの存在——これらを理解した上で、プロジェクトのどの部分を非同期化するか、どの部分を同期のまま残すかを判断することが、Django ASGI を実務で活用するための前提です。