ドキュメント価格ブログ
お問い合わせ
ログイン無料で新規登録
Back to Tech Articles

FastAPIとDjango Channelsによるリアルタイム通信：Websocketsの実装

8月 03, 2025
# Misc

Takashi Yamamoto

Infrastructure Engineer · Leapcell

この記事はEnglish、Deutschでも利用できます
Cover of "FastAPIとDjango Channelsによるリアルタイム通信：Websocketsの実装"

今日のペースの速いデジタル世界では、即時の情報交換に対する期待はかつてないほど高まっています。共同ドキュメント編集やライブチャットアプリケーションから、金融取引プラットフォームやリアルタイムダッシュボードまで、動的で即時更新の需要は遍在しています。伝統的なHTTPリクエスト-レスポンスサイクルは、基本的ではありますが、このシームレスなリアルタイム体験を提供するという点ではしばしば不十分です。ここでWebSocketsが強力な代替手段として登場し、クライアントとサーバー間の永続的で双方向の通信チャネルを可能にします。FastAPIやDjangoのような最新のPython WebフレームワークでWebSocketsを効果的に活用する方法を理解することは、次世代のインタラクティブアプリケーションを構築するために不可欠です。この記事では、FastAPIとDjango Channelsの両方でWebSocket接続を実装するためのメカニズムを探り、プロジェクトにリアルタイム機能統合するための知識と例を提供します。

リアルタイム通信の基本を理解する

実装の詳細に入る前に、リアルタイムWebアプリケーションの基盤となるコアコンセプトを明確に理解しましょう。

WebSockets:"HTTPとは異なり、ステートレスで各情報に個別のリクエストを必要とするのに対し、WebSocketは単一の、長時間持続するTCP接続を介してフルデュプレックス通信チャネルを提供します。これは、クライアントとサーバーの両方が、各交換で新しい接続を確立するオーバーヘッドなしに同時にデータを送受信できることを意味し、大幅に低いレイテンシーとより効率的なリソース利用につながります。接続はHTTPハンドシェイクによって開始され、その後WebSocketプロトコルに「アップグレード」されます。

ASGI (Asynchronous Server Gateway Interface): ASGIはWSGI (Web Server Gateway Interface) の精神的な後継であり、非同期操作を処理するために設計されています。WSGIは同期型で従来の要求-応答アプリケーションに適していますが、ASGI はPythonのWebサーバー機能を拡張して、長時間接続、WebSocket、およびロングポーリングを組み込みます。FastAPIとDjango Channelsの両方がASGIの上に構築されており、非同期性を活用して多数の同時WebSocket接続を効率的に管理します。

Channels (Django Channels): Djangoはデフォルトで、従来のHTTPリクエストのために設計された同期フレームワークです。Django Channels は、HTTP以外のプロトコル、WebSocket、チャットプロトコル、IoTプロトコルを処理するためにDjangoの機能を拡張します。これには3つの主要コンポーネントが含まれます: プロセス間通信のためのChannel Layers、接続を処理するためのConsumers、および着信接続を適切なコンシューマーにルーティングするためのRoutersです。

FastAPI: Python 3.7+ 用のAPI構築のための、モダンで高速（高性能）なWebフレームワークで、標準的なPython型ヒントに基づいています。FastAPIは本質的に非同期操作をサポートしており、 WebSocket のファーストクラスサポートを提供するため、リアルタイム機能の統合が非常に簡単です。

リアルタイムアプリケーションの構築

次に、FastAPIとDjango ChannelsでWebSocket接続を実装する方法を、実践的なコード例とともに見ていきましょう。

FastAPIでのWebSocket実装

FastAPIの非同期性は、WebSocketの統合を驚くほど直観的にします。HTTPエンドポイントを定義するのと同様の方法で、WebSocketエンドポイントを定義します。

# main.py
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List

app = FastAPI()

# 実際には、ブロードキャストのためにインメモリまたはRedisのpub/subを使用する可能性が高いです。
# このシンプルなリストは、デモ目的で接続されているクライアントをシミュレートします。
connected_clients: List[WebSocket] = []

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
    await websocket.accept()
    connected_clients.append(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            print(f"Received message from client {client_id}: {data}")
            # すべての接続されたクライアントにメッセージをブロードキャスト
            for client in connected_clients:
                if client != websocket:  # 送信者へのエコーを避ける
                    await client.send_text(f"Client {client_id} says: {data}")
                else:
                    await client.send_text(f"You said: {data}")
    except WebSocketDisconnect:
        connected_clients.remove(websocket)
        print(f"Client {client_id} disconnected.")
        # 切断を他のクライアントに通知
        for client in connected_clients:
            await client.send_text(f"Client {client_id} has disconnected.")

# これを実行するには、main.pyとして保存し、以下を実行します: uvicorn main:app --reload

説明:

  1. @app.websocket("/ws/{client_id}"): このデコレーターはWebSocketエンドポイントを登録します。{client_id}はパスパラメータです。
  2. websocket: WebSocket: FastAPIは自動的にWebSocketオブジェクトを注入し、接続を操作できます。
  3. await websocket.accept(): これは不可欠です。WebSocketハンドシェイクを完了し、永続的な接続を確立します。これがないと、接続は開きません。
  4. connected_clients.append(websocket): クライアント間のメッセージブロードキャストを容易にするために、接続されているクライアントのリストを維持します。
  5. while True: data = await websocket.receive_text(): このループは、クライアントからの着信テキストデータを継続的にリッスンします。receive_text()は非同期関数です。
  6. await client.send_text(...): これは、接続されているクライアントにテキストデータを送信します。
  7. except WebSocketDisconnect: クライアントが切断されたときにこの例外がスローされます。クライアントをconnected_clientsリストから削除するなど、クリーンアップロジックを処理するためのクリーンな方法です。

クライアント側 (JavaScript):

<!-- index.html -->
<!DOCTYPE html>
<html>
<head>
    <title>FastAPI WebSocket Chat</title>
</head>
<body>
    <h1>FastAPI WebSocket Chat</h1>
    <input type="text" id="messageInput" placeholder="Type your message">
    <button onclick="sendMessage()">Send</button>
    <ul id="messages"></ul>

    <script>
        const clientId = Math.floor(Math.random() * 1000); // デモ用の簡単なID
        const socket = new WebSocket(`ws://localhost:8000/ws/${clientId}`);

        socket.onopen = function(event) {
            console.log("WebSocket connection opened:", event);
            addMessage("You are connected!");
        };

        socket.onmessage = function(event) {
            console.log("Message from server:", event.data);
            addMessage(event.data);
        };

        socket.onclose = function(event) {
            console.log("WebSocket connection closed:", event);
            addMessage("Disconnected from server.");
        };

        socket.onerror = function(error) {
            console.error("WebSocket error:", error);
            addMessage("WebSocket error occurred.");
        };

        function sendMessage() {
            const input = document.getElementById("messageInput");
            const message = input.value;
            if (message) {
                socket.send(message);
                input.value = "";
            }
        }

        function addMessage(text) {
            const ul = document.getElementById("messages");
            const li = document.createElement("li");
            li.textContent = text;
            ul.appendChild(li);
        }
    </script>
</body>
</html>

このシンプルなHTMLページは、WebSocket接続を開始し、基本的なチャットインターフェースを提供します。

Django ChannelsでのWebSocket実装

Django Channels は、そのモジュール設計により、FastAPIよりも少し多くのセットアップが必要ですが、特にDjangoのORMや認証システムと組み合わせて、複雑なリアルタイムアプリケーションのための堅牢な機能を提供します。

1. Channelsのインストール:

pip install channels

2. settings.pyにchannelsをINSTALLED_APPSに追加:

# myproject/settings.py
INSTALLED_APPS = [
    # ... other apps
    'channels',
    'chat', # あなたのアプリが'chat'という名前だと仮定します
]

3. ASGIアプリケーションを定義 (myproject/asgi.py):

# myproject/asgi.py
import os

from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from django.core.asgi import get_asgi_application

import chat.routing # あなたのアプリのルーティングをインポートします

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "myproject.settings")

application = ProtocolTypeRouter({
    "http": get_asgi_application(), # 標準のHTTPリクエストを処理します
    "websocket": AuthMiddlewareStack(
        URLRouter(
            chat.routing.websocket_urlpatterns
        )
    ),
})

4. WebSocketsのルーティングを定義 (chat/routing.py):

このファイルはURLパスとWebSocketコンシューマーをマッピングします。

# chat/routing.py
from django.urls import re_path

from . import consumers

websocket_urlpatterns = [
    re_path(r"ws/chat/(?P<room_name>\w+)/$", consumers.ChatConsumer.as_asgi()),
]

5. WebSocketコンシューマーを作成 (chat/consumers.py):

コンシューマーは、プロトコルの種類に固有のイベントを処理する非同期関数です。WebSocketの場合、AsyncWebsocketConsumerから継承します。

# chat/consumers.py
import json

from channels.generic.websocket import AsyncWebsocketConsumer

class ChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.room_name = self.scope["url_route"]["kwargs"]["room_name"]
        self.room_group_name = "chat_" % self.room_name

        # ルームグループに参加
        await self.channel_layer.group_add(
            self.room_group_name,
            self.channel_name
        )
        await self.accept()

    async def disconnect(self, close_code):
        # ルームグループを退出
        await self.channel_layer.group_discard(
            self.room_group_name,
            self.channel_name
        )

    # WebSocketからのメッセージ受信
    async def receive(self, text_data):
        text_data_json = json.loads(text_data)
        message = text_data_json["message"]

        # ルームグループにメッセージを送信
        await self.channel_layer.group_send(
            self.room_group_name,
            {
                "type": "chat.message", # 以下のchat_messageメソッドを呼び出します
                "message": message
            }
        )

    # ルームグループからのメッセージ受信
    async def chat_message(self, event):
        message = event["message"]

        # WebSocketにメッセージを送信
        await self.send(text_data=json.dumps({"message": message}))

説明:

  • AsyncWebsocketConsumer: connect, disconnect, receiveのような基本的なWebSocketメソッドを提供します。
  • self.scope: URLパラメータを含む、接続に関する情報が含まれています。
  • channel_layer: これはChannelsのプロセス間通信の中核です。これにより、異なるコンシューマー（異なるサーバープロセス間でも）が互いに通信できます。ここでは、コンシューマーを「グループ」（チャットルーム）に追加し、そのグループにメッセージを送信するために使用します。
  • group_add, group_discard: コンシューマーをグループに追加/削除するためのメソッドです。
  • group_send: 指定されたグループ内のすべてのコンシューマーにメッセージを送信します。"type": "chat.message" は、そのグループ内のすべてのコンシューマーでchat_messageメソッドを呼び出すようにChannelsに指示します。
  • self.send(): WebSocket経由でクライアントにデータを送信します。

6. Channel Layersの設定 (settings.py):

スケーリングのためには、Channel Layerのバックエンド（例: Redis）が必要になります。シンプルな開発セットアップには、インメモリChannel Layerで十分です。

# myproject/settings.py
CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels.layers.InMemoryChannelLayer"
        # 本番環境用: "BACKEND": "channels_redis.core.RedisChannelLayer",
        # "CONFIG": {
        #     "hosts": [("127.0.0.1", 6379)],
        # },
    },
}

7. チャットルームテンプレート用の基本的なDjangoビューを作成 (chat/views.py):

# chat/views.py
from django.shortcuts import render

def room(request, room_name):
    return render(request, "chat/room.html", {"room_name": room_name})

8. プロジェクトURLにチャットアプリのURLを含める (myproject/urls.py):

# myproject/urls.py
from django.contrib import admin
from django.urls import include, path

urlpatterns = [
    path("admin/", admin.site.urls),
    path("chat/", include("chat.urls")), # チャットアプリのURLを含める
]

9. チャットルームテンプレートを作成 (chat/templates/chat/room.html):

<!-- chat/templates/chat/room.html -->
<!DOCTYPE html>
<html>
<head>
    <title>Django Channels Chat</title>
</head>
<body>
    <h1>Chat Room: {{ room_name }}</h1>
    <textarea id="chat-log" cols="100" rows="20"></textarea><br>
    <input id="chat-message-input" type="text" size="100"><br>
    <input id="chat-message-submit" type="button" value="Send">

    <script>
        const roomName = JSON.parse(document.getElementById('room-name').textContent);
        const chatLog = document.getElementById('chat-log');
        const messageInput = document.getElementById('chat-message-input');
        const messageSubmit = document.getElementById('chat-message-submit');

        const chatSocket = new WebSocket(
            'ws://'
            + window.location.host
            + '/ws/chat/'
            + roomName
            + '/'
        );

        chatSocket.onopen = function(e) {
            chatLog.value += "Connected to chat room.\n";
        };

        chatSocket.onmessage = function(e) {
            const data = JSON.parse(e.data);
            chatLog.value += (data.message + '\n');
        };

        chatSocket.onclose = function(e) {
            console.error('Chat socket closed unexpectedly');
            chatLog.value += "Disconnected from chat room.\n";
        };

        messageInput.focus();
        messageInput.onkeyup = function(e) {
            if (e.keyCode === 13) {  // enter, return
                messageSubmit.click();
            }
        };

        messageSubmit.onclick = function(e) {
            const message = messageInput.value;
            chatSocket.send(JSON.stringify({
                'message': message
            }));
            messageInput.value = '';
        };
    </script>
    <textarea id="room-name" style="display: none;">{{ room_name }}</textarea>
</body>
</html>

Django Channelsの実行:

python manage.py runserver

その後、ブラウザでhttp://localhost:8000/chat/myroom/にアクセスします。

アプリケーションシナリオ

FastAPIとDjango Channelsは、リアルタイム更新が必要なシナリオで優れています。

  • ライブチャット: 例で示したように、インスタントメッセージングアプリケーションに最適です。
  • リアルタイムダッシュボード: ページリフレッシュなしで、ライブ分析、株価、またはセンサーデータを表示します。
  • 共同作業ツール: 複数のユーザーが同じコンテンツを同時に操作する共有ホワイトボード、ドキュメント編集（例: Google Docs）。
  • ゲーム: プレイヤーの位置やゲーム状態の即時更新が重要なシンプルなマルチプレイヤーゲーム。
  • 通知: ユーザーに即時通知をプッシュします（例: 新しいメッセージ、友達リクエスト）。

結論

FastAPIとDjango Channelsを使用したPythonでのWebSocket接続の実装は、非常にインタラクティブで応答性の高いWebアプリケーションを構築するための可能性の世界を開きます。FastAPIは、ネイティブの非同期サポートを活用して迅速なセットアップを行うための、合理化された「すべて込み」のWebSocketアプローチを提供します。Django Channelsは、より多くの初期設定が必要ですが、Djangoエコシステム全体で複雑なリアルタイム機能のための堅牢でスケーラブルなアーキテクチャを提供し、既存のDjangoプロジェクトとのシームレスな統合と、その強力なORMおよび認証の活用を可能にします。どちらのフレームワークも、効果的にASGIのパワーを活用して効率的で双方向の通信を提供し、開発者がユーザーをリアルタイムで真に接続する最新のWebエクスペリエンスを構築できるようにします。

Share this article

HackerNews

More Posts from Leapcell

cover

Gunicorn, Uvicorn、Nginx を使用した Python Web アプリケーションのプロダクションデプロイ

cover

PydanticとFastAPIによるAPIドキュメントの効率化

Popular Posts