5. Django は ASGI にどう対応しているか

5.1. asgi.py

Django プロジェクトを django-admin startproject myproject で生成すると、wsgi.py と並んで asgi.py が自動的に作成されます。

# 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()

この構造は1 章（Django を WSGI 視点で見る）で見た wsgi.py とほぼ同一です。

# 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 での運用に戻せます。

5.2. get_asgi_application()

get_asgi_application() の内部を追うと、Django の ASGI 対応がどのように実装されているかが見えてきます。

# 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 に定義されており、2 章（なぜ ASGI が必要になったのか）で学んだ ASGI の3引数インタフェースを実装しています。

# 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"] による分岐

2.5 章（ASGI の基本構造）で学んだ通り、ASGI アプリケーションは http、websocket、lifespan の3種のスコープを処理する必要があります。Django の ASGIHandler は http と lifespan に対応しますが、websocket は標準では対応していません。WebSocket を扱う場合は Django Channels を導入します。

2. read_body メソッド

3.4 章（request body を受け取る）で手書きした read_body(receive) ヘルパーと本質的に同じ処理が Django 内部にも存在しており、more_body フラグを監視しながらループして全ボディを蓄積します。

3. ASGIRequest の生成

1.3 章（リクエストは Django にどう渡るか）で WSGIRequest が environ 辞書からリクエスト属性を構築したように、ASGIRequest は scope 辞書と受信済みボディからリクエスト属性を構築します。

# 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 の処理は、2.6 章（scope を理解する）で見た scope のキーを1.3 章（リクエストは Django にどう渡るか）で見た environ のキー形式（HTTP_* プレフィックス付き大文字）に逆変換しています。Django の内部コード（ミドルウェア、認証、CSRF 検証など）はすべて request.META["HTTP_HOST"] や request.META["CONTENT_TYPE"] の形式を前提としているため、ASGI の scope["headers"]（小文字バイト列タプル）をこの形式に合わせる必要があります。

注釈

この変換層が存在することで、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) は1.5 章（middleware chain の流れ）で見たミドルウェアチェーンを通じてビュー関数を実行し、HttpResponse を返します。このレスポンスオブジェクトに対して await response(scope, receive, send) を呼ぶことで、ASGI のイベント形式に変換されます。

# 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})

3 章（最小の ASGI HTTP アプリ）で手書きした send({"type": "http.response.start", ...}) と send({"type": "http.response.body", ...}) の2段階送信が、Django の HttpResponse にも組み込まれています。1.7 章（レスポンス生成）で見た 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)

重要

ASGIHandler.handle から get_response が呼ばれた後は、1 章（Django を WSGI 視点で見る）で追った WSGI 時のフロー（ミドルウェアチェーン、URL 解決、ビュー実行、レスポンス生成）がまったく同じように動作します。ASGI 対応によって変わったのは入口（scope / receive → ASGIRequest）と出口（HttpResponse → send イベント）の変換層だけであり、Django の中核ロジックは WSGI / ASGI に対して透過的に設計されています。

次節では、Django の async def ビューの仕組みと、同期ビューが ASGI 上でどのように実行されるかを詳しく追います。

5.3. Django の request 処理と async

5.3.1. 同期 view

Django の伝統的なビュー関数は、同期関数として定義されます。

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 クエリが完了するまでそのスレッドがブロックされ、レスポンスを返します。1 章（Django を WSGI 視点で見る）で追った WSGI フローそのものです。

問題は、ASGI サーバ（Uvicorn / Daphne）上で同期ビューが呼ばれた場合です。

• ASGI サーバはイベントループ上で await ASGIHandler.__call__(scope, receive, send) を実行しています

• もし同期ビューをイベントループ上で直接呼び出すと、User.objects.get() がデータベース応答を待つ間、イベントループ全体がブロックされます

• これは 4.10 章（トラブルシューティングの観点）で見た「async def 内での blocking I/O」と同じ問題です

Django はこの問題を自動的に解決します。ASGIHandler はビュー関数が同期関数かどうかを asyncio.iscoroutinefunction() で検査し、同期関数であればスレッドプールにオフロードして実行します。

# 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 で実行するのと同じ設計判断です（4 章（FastAPI を ASGI 視点で見る））。

ミドルウェアについても同様の処理が行われます。Django のミドルウェアは歴史的に同期で書かれているものが大半です。ASGIHandler はミドルウェアチェーンの各ミドルウェアが同期か非同期かを判定し、同期ミドルウェアはスレッドプールで実行します。この判定と切り替えの処理が、ASGI 上で既存の同期コードをそのまま動作させる互換層として機能しています。

5.3.2. 非同期 view

Django 4.1 以降では、ビュー関数を async def で定義できるようになりました。

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 など）をスレッドプールで呼び出す形です。

# Django ORM の非同期メソッドの内部（概念コード）
class QuerySet:
    async def aget(self, **kwargs):
        return await sync_to_async(self.get)(**kwargs)

注釈

「非同期ビューを書いてもデータベースアクセスが本質的に非同期になるわけではない」という点に注意してください。パフォーマンス上の利点は、イベントループがブロックされないことにより他のリクエストを並行処理できる点にあります。真の非同期データベースアクセスが必要な場合は、encode/databases や SQLAlchemy の async 拡張を Django と組み合わせることになります。

非同期ビュー内で同期 ORM メソッドを直接呼ぶと、Django は SynchronousOnlyOperation 例外を送出します。

async def bad_view(request, user_id):
    user = User.objects.get(id=user_id)  # SynchronousOnlyOperation!
    return JsonResponse({"id": user.id})

重要

この例外は安全弁として機能しています。async def 内で同期ブロッキング呼び出しが行われることを検出し、明示的にエラーにすることで、4.10 章（トラブルシューティングの観点）で述べたイベントループブロッキングの問題を未然に防ぎます。FastAPI では同じ問題が警告なしに発生するのに対し、Django はフレームワークレベルで検出する仕組みを持っています。

sync_to_async デコレータを使えば、任意の同期関数を非同期ビュー内から安全に呼び出せます。

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 です。

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)

5.3.3. 実行経路の違い

同期ビューと非同期ビューの実行経路を並べると、Django が ASGI 上で2つの異なるパスを持っていることがわかります。

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 ヘビーなビュー

注意

sync_to_async と async_to_sync のブリッジは便利ですが、呼び出しのたびにスレッドプールのオーバーヘッドが発生します。頻繁な切り替えはパフォーマンスに影響するため、境界を越える回数を最小限に抑えることを意識してください。

次節では Django Channels を通じた WebSocket 対応と、Django の ASGI 拡張の全体像を確認します。

5.4. sync / async 境界の橋渡し

5.4.1. asgiref

Django の同期・非同期変換は asgiref というライブラリが担っています。asgiref は Django プロジェクトが開発・保守しているパッケージで、Django をインストールすると依存関係として自動的にインストールされます。

asgiref は ASGI 仕様のリファレンス実装とユーティリティを提供するライブラリですが、その中で最も重要な機能が sync_to_async と async_to_sync の2つのブリッジ関数です。これらは Django 内部のあらゆる場所で使われており、5.3 章（Django の request 処理と async）で見た「同期ビューを ASGI 上でスレッドプール実行する」仕組みも、「非同期ビューを WSGI 上で実行する」仕組みも、すべて asgiref を通じて実現されています。

# asgiref は Django と一緒にインストール済み
from asgiref.sync import sync_to_async, async_to_sync

asgiref が解決する問題を理解するには、Python の asyncio における基本的な制約を知る必要があります。

• asyncio のイベントループは単一スレッドで動作し、await 可能なコルーチンだけを扱います

• 同期関数をイベントループ上で直接呼ぶとブロッキングが発生します（2.11 章（トラブルシューティングの観点）、4.10 章（トラブルシューティングの観点））

• 逆にイベントループが走っていない同期コンテキストではコルーチンを await できません

asgiref はこの双方向の壁を、スレッドプールとイベントループの生成・管理によって透過的に越える仕組みを提供しています。

5.4.2. sync_to_async

sync_to_async は、同期関数を非同期コンテキストから安全に呼び出すためのラッパーです。

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 パラメータです。

# 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 に適しています。

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) が使われています。5.3 章（Django の request 処理と async）で示した実行経路の図で「スレッドプール実行」としていた部分の正体がこれです。

# Django 内部での使用例（概念コード）
# django/core/handlers/base.py
if not asyncio.iscoroutinefunction(middleware):
    middleware = sync_to_async(middleware, thread_sensitive=True)

デコレータとしても使えます。

@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})

5.4.3. async_to_sync

async_to_sync は反対方向の変換で、非同期関数を同期コンテキストから呼び出せるようにするラッパーです。

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 が使われています。

# django/core/handlers/base.py（概念コード）
if asyncio.iscoroutinefunction(callback):
    # WSGI 上で async ビューを実行する場合
    callback = async_to_sync(callback)
response = callback(request, *args, **kwargs)

5.3 章（Django の request 処理と async）で「WSGI サーバ上で async def ビューを実行した場合、動作はするが非同期の利点がない」と述べた理由がここにあります。async_to_sync はコルーチンを実行完了まで同期的にブロックするため、await による並行処理の恩恵は得られません。

async_to_sync が最も有用なのは、Django の管理コマンドやテスト、Celery タスクなど、同期的な実行環境から非同期ライブラリを利用する場合です。

# 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 はイベントループの作成・管理にコストがかかります。ループ内で毎回変換を繰り返すと顕著な遅延が生じます。

# 悪い例: ループ内で毎回 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) を呼ぶと、単一スレッドが既に占有されているためデッドロックします。

# デッドロックの可能性がある例
@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 がデフォルトである理由の一つがこれです。

全体の関係を図にすると次のようになります。

同期の世界                           非同期の世界
──────────────                     ──────────────
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 拡張の全体像を確認します。

5.5. middleware は async にどう対応するか

5.5.1. sync middleware

Django のミドルウェアは、歴史的にすべて同期で書かれてきました。1.5 章（middleware chain の流れ）で見た標準的なミドルウェアの構造を振り返ります。

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 はミドルウェアチェーンを構築する段階で各ミドルウェアが同期か非同期かを検査します。同期ミドルウェアが検出された場合、5.4 章（sync / async 境界の橋渡し）で見た sync_to_async を使って非同期呼び出し可能な形にラップします。

# 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 互換になります。

注釈

5.4 章（sync / async 境界の橋渡し）で述べた通り、sync_to_async(thread_sensitive=True) はすべてのリクエストで同じ単一スレッドを使います。ミドルウェアチェーン内の同期ミドルウェアが多いほど、そのスレッドがボトルネックになります。リクエスト処理のうちミドルウェア通過部分だけが直列化され、非同期の並行処理の利点が減少します。

5.5.2. async middleware

Django 4.1 以降では、ミドルウェアを非同期で定義できるようになりました。非同期ミドルウェアは ASGI サーバのイベントループ上で直接実行され、スレッドプールへのオフロードが発生しません。

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 を使えることです。

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 のクラス属性を検査し、実行環境に応じた呼び出し方を選択します。

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 など）もこのデュアルモードパターンに順次対応しつつあります。

5.5.3. 混在時の考え方

実務のプロジェクトでは、同期ミドルウェアと非同期ミドルウェアが混在する状況が一般的です。

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 でラップします。チェーン全体の実行の流れを追うと次のようになります。

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 の非同期対応状況に依存します

パフォーマンスへの影響を測定するには、5.3 章（Django の request 処理と async）で示したように各ミドルウェアの前後で時間を計測します。

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 ミドルウェア対応は「既存の同期コードを壊さずに動かす」ことを最優先にしつつ、「新しいコードは非同期で書ける」余地を提供する段階的移行戦略です。3.6 章（ASGI ミドルウェアを書く）で手書きした ASGI ミドルウェアが async def __call__(self, scope, receive, send) で send をラップしていたのに対し、Django のミドルウェアは request / response オブジェクトを介した高レベルの抽象化を維持しています。これにより開発者は scope や receive / send を直接扱う必要がなく、同期・非同期の切り替えも def __call__ を async def __call__ に変えるだけで済みます。

次節では Django Channels による WebSocket 対応を見ていきます。

5.6. Django で ASGI を使う場面

5.6.1. async view

5.3 章（Django の request 処理と async）で非同期ビューの仕組みを追いましたが、ここでは「どのようなケースで async def ビューを選択すべきか」という実務上の判断基準を整理します。

非同期ビューが効果を発揮するのは、ビュー関数内で複数の独立した I/O 待ちが発生するケースです。

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つのスレッドで実行するため、関数内部の並行化は行われません。

一方で、単一の ORM クエリを実行して結果を返すだけのビューでは、非同期にする利点はほとんどありません。

Tip

def ビューと async def ビューを使い分ける目安: 外部 I/O が1つだけなら def、複数の外部 I/O を並行して待つ場合は async def を検討してください。

# このビューを 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 待ちで完了し、並行化の余地がありません。5.3 章（Django の request 処理と async）で述べた通り、Django の非同期 ORM は内部的に sync_to_async でラップされているため、await User.objects.aget(id=user_id) としてもデータベースドライバレベルでは同期のままです。このケースでは def のまま残し、ASGI 上では Django が自動的にスレッドプールで実行する方がシンプルです。

5.6.2. 長時間待ち I/O

非同期ビューが特に有効なのは、レスポンスの生成に長い I/O 待ちを伴うケースです。

Server-Sent Events（SSE）によるストリーミングはその典型的な例です。

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 のトークンストリーミングも同様のパターンです。

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 呼び出しも、非同期ビューが効果的な場面です。

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件の応答時間程度で完了します。

5.6.3. WebSocket は素の Django だけでは足りないこと

5 章（Django は ASGI にどう対応しているか）で見た通り、Django の ASGIHandler は scope["type"] == "http" と scope["type"] == "lifespan" を処理しますが、scope["type"] == "websocket" には対応していません。

警告

素の Django（Channels なし）で WebSocket 接続を受け付けると、ASGIHandler が ValueError: Unknown scope type: websocket を送出します。WebSocket が必要な場合は Django Channels を導入してください。

# 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 が来るとここで例外

2.9 章（WebSocket と lifespan）と3.7 章（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 のコアに組み込むには大幅な設計変更が必要です。

5.6.4. Channels の位置づけを軽く紹介

Django Channels は、Django に WebSocket と他の非 HTTP プロトコルのサポートを追加するための公式拡張パッケージです。Django プロジェクトの一部として開発されていますが、Django 本体には含まれていません。

pip install channels

Channels の核心的なアイデアは、Django の ASGIHandler を別の ASGI アプリケーション（ProtocolTypeRouter）でラップし、scope["type"] に応じて処理を振り分けることです。

注釈

ProtocolTypeRouter は3.5 章（ルーティングを自作してみる）で自作したルーターの scope["type"] 分岐と同じ発想です。HTTP リクエストは従来通り Django の ASGIHandler が処理し、WebSocket 接続は Channels の URLRouter が処理します。

# 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)
    ),
})

# myapp/routing.py
from django.urls import path
from myapp.consumers import ChatConsumer

websocket_urlpatterns = [
    path("ws/chat/<str:room_name>/", ChatConsumer.as_asgi()),
]

# 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 は3.7 章（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 機構です。3.7 章（WebSocket の最小実装）のチャットルーム実装では connected 集合にインメモリで send callable を保持しましたが、これは単一プロセスでしか動作しません。Channel Layer はプロセスの壁を越えてメッセージを配信します。

本書では Channels の詳細な実装には踏み込みませんが、重要なのは Channels が Django のコアを変更するのではなく、ASGI のプロトコルルーティングを活用して Django の横に WebSocket 処理を並置する設計であるという点です。Django 本体は依然として HTTP 処理に集中し、WebSocket は Channels が受け持つという責務分離が、ASGI の scope["type"] 分岐によって実現されています。

次節では Django の ASGI 対応に関するトラブルシューティングの観点を整理します。

5.7. Django ASGI の限界と誤解しやすい点

5.7.1. 「ASGI 対応」と「全部 async で速い」は違う

Django を ASGI サーバ上で動かした瞬間にすべてが非同期になり高速化されるという期待は、最もよくある誤解です。

注意

uvicorn myproject.asgi:application で起動しただけでは、既存の同期ビューと同期ミドルウェアの実行方式が変わるだけです。「ASGI にすれば速くなる」という期待は誤りです。

5.3 章（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 の利点が発揮されるのは、5.6 章（Django で ASGI を使う場面）で述べた「複数の外部 API を並行呼び出しする」「SSE やストリーミングで長時間接続を保持する」「WebSocket を使う」といった、非同期 I/O が本質的に必要な場面に限られます。

ASGI へ移行すべきかどうかの判断基準を整理します。

状況	推奨
非同期 I/O が必要なエンドポイントがある	ASGI へ移行
WebSocket や SSE の要件がある	ASGI へ移行
将来的に非同期化を段階的に進める計画がある	ASGI へ移行
ORM 中心の CRUD アプリケーション	WSGI のまま運用
管理画面が主体のシステム	WSGI のまま運用
サードパーティパッケージが同期前提のプロジェクト	WSGI のまま運用

5.7.2. ORM と async の注意点

Django ORM の非同期対応は進行中ですが、現時点ではいくつかの重要な制約があります。

5.3 章（Django の request 処理と async）で述べた通り、非同期 ORM メソッド（aget, acreate, asave など）の多くは内部的に sync_to_async で同期版をラップしています。データベースドライバ（psycopg2, mysqlclient など）自体が同期のため、await User.objects.aget(id=1) はスレッドプール内で User.objects.get(id=1) を実行し、その完了を await で待つ構造です。

重要

真の非同期データベースアクセス（ドライバレベルでの non-blocking I/O）は実現されていません。非同期 ORM に切り替えたからといってデータベースアクセスが高速化されるわけではなく、利点はイベントループがブロックされないことにあります。

# これらは見た目は非同期だが、内部的にはスレッドプール経由の同期実行
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 例外が発生します。これは意図的な安全機構ですが、実務では思わぬ場所でこの例外に遭遇します。

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 でラップする

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 の構築自体は同期でも非同期でも安全ですが、評価のタイミングには注意が必要です。

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 を使ってレコードを取得します。

5.7.3. middleware やサードパーティパッケージの対応状況

Django エコシステムの大部分は同期を前提に構築されています。ASGI に移行する際、自分のコード以外の部分が問題の原因になることがあります。

Django 自身の組み込みミドルウェアはバージョンアップのたびにデュアルモード（同期・非同期両対応）化が進んでいます。5.5 章（middleware は async にどう対応するか）で見た通り、SecurityMiddleware、SessionMiddleware、CommonMiddleware、CsrfViewMiddleware、AuthenticationMiddleware などは非同期対応が進んでいます。しかし、これらのミドルウェアが内部で ORM にアクセスする際（セッションの読み書き、ユーザーの取得など）は sync_to_async を経由するため、完全な非同期実行にはなっていません。

サードパーティパッケージの対応状況はまちまちです。

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',                         # 自分次第
]

同期のみのサードパーティミドルウェアは、5.5 章（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 ゲートウェイやリバースプロキシで結合するという構成も現実的な選択肢です

Django ASGI の現在地

Django の ASGI 対応は「すべてを一度に非同期化する」ことを目指していません。Django のコア開発チームは DEP（Django Enhancement Proposal）を通じて段階的な非同期化ロードマップを示しており、ORM の完全非同期化、ミドルウェアの完全非同期化、テンプレートレンダリングの非同期化などが将来的に計画されています。現時点での Django ASGI は「完成した非同期フレームワーク」ではなく「同期から非同期へ移行する途中の状態」であり、その制約を理解した上で使うことが重要です。

次節では、本章のまとめとして Django の ASGI 対応全体を振り返り、トラブルシューティングの観点で整理します。

5.8. トラブルシューティングの観点

5.8.1. 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 呼び出しです。

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 に書き換えただけでは解決しないケースがあります。5.7 章（Django ASGI の限界と誤解しやすい点）で述べた遅延読み込みの問題です。

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つあります。

# パターン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() でイベントループの有無を検査し、イベントループ内であれば同期操作を拒否する仕組みを持っています。

5.8.2. event loop まわりの誤解

ASGI 環境でのイベントループに関する誤解は、デバッグを困難にする原因になります。

最も多い誤解は「asyncio.run() で非同期コードを実行すればよい」というものです。

危険

ASGI サーバ上の同期ビュー内で asyncio.run() を使ってはいけません。Python の asyncio は同一スレッドに複数のイベントループを許容しないため、RuntimeError: This event loop is already running が発生します。代わりに async_to_sync を使ってください。

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 を使うことです。

from asgiref.sync import async_to_sync

def correct_view(request):
    result = async_to_sync(fetch_data)()
    return JsonResponse(result.json())

async_to_sync は現在のスレッドにイベントループが存在する場合としない場合の両方を正しく処理します。5.4 章（sync / async 境界の橋渡し）で述べた通り、イベントループが存在する場合は別スレッドで実行し、存在しない場合は新しいイベントループを作成します。

もう一つの誤解は、asyncio.get_event_loop() を使ってタスクを手動でスケジュールしようとするケースです。

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 環境変数に関する誤解もあります。

# 開発時のシェルで見かける設定
os.environ["DJANGO_ALLOW_ASYNC_UNSAFE"] = "true"

この環境変数は SynchronousOnlyOperation のチェックを無効化します。Django シェル（manage.py shell）や Jupyter Notebook のように、イベントループが自動的に作成される環境で同期 ORM を使いたい場合の回避策です。

危険

DJANGO_ALLOW_ASYNC_UNSAFE=true を本番環境に設定してはいけません。非同期ビュー内の同期 ORM 呼び出しが例外にならず、イベントループのブロッキングが黙って発生します。2.11 章（トラブルシューティングの観点）や 4.10 章（トラブルシューティングの観点）で述べた「async 内で blocking」の問題がフレームワークの安全弁なしに起きます。

5.8.3. async view に blocking 処理を入れる問題

SynchronousOnlyOperation は Django が検出できるブロッキングのみを対象としています。Django の管理下にない同期ブロッキング呼び出しは検出されず、黙ってイベントループを停止させます。

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リクエストずつ処理する限り、ブロッキングの影響は自分自身のレスポンスタイムに限定されます

• 同時リクエスト数が増えて初めて、他のリクエストのレスポンスタイムが連鎖的に悪化します

検出方法として最も有効なのは、5.4 章（sync / async 境界の橋渡し）でも触れた PYTHONASYNCIODEBUG=1 環境変数です。

PYTHONASYNCIODEBUG=1 uvicorn myproject.asgi:application --port 8000

この設定により、イベントループ上で100ms以上ブロックする処理が警告として出力されます。開発環境で有効にしておけば、ブロッキング呼び出しを早期に発見できます。

もう一つの検出方法は、ミドルウェアでリクエスト処理時間を計測し、異常に長い処理を記録することです。

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

ブロッキングを発見した場合の対処は、4.10 章（トラブルシューティングの観点）で述べた3つのパターンと同じです。

1. 非同期ライブラリへの置き換え（最善）: requests → httpx、time.sleep → asyncio.sleep、open → aiofiles

2. スレッドプールへの逃避: 置き換えが困難な場合は sync_to_async(thread_sensitive=False) でスレッドプールに逃がします

3. def ビューに変更: そもそも async def にする必要がないビューであれば def に変更し、Django の自動スレッドプール実行に委ねます

重要

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 を実務で活用するための前提です。