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

FastAPIとWebSocketsによるPythonでのリアルタイム通信

8月 15, 2025
# Misc

Grace Collins

Solutions Engineer · Leapcell

この記事はEnglish、한국어、Deutschでも利用できます
Cover of "FastAPIとWebSocketsによるPythonでのリアルタイム通信"

はじめに

今日の相互接続された世界では、インスタントアップデートとダイナミックなユーザーエクスペリエンスへの需要はかつてないほど高まっています。共同ドキュメント編集からライブチャットアプリケーションまで、リアルタイムで通信できる能力は、最新のWebアプリケーションにとって不可欠です。堅牢な多くのタスクにおいて、従来のHTTPリクエスト・レスポンスモデルは、継続的で双方向の通信が必要な場合にはしばしば不十分です。そこにWebSocketのようなテクノロジーが登場し、サーバーが利用可能になり次第、クライアントに情報をプッシュできるようにする（またはその逆）永続的な接続を提供します。Pythonは、その強力なエコシステムとフレームワークにより、このようなリアルタイムシステムを構築するための優れたツールを提供します。この記事では、Pythonでのリアルタイム通信の実装、特にWebSocketを活用する方法を掘り下げ、FastAPIがこのプロセスをどのように簡略化し、開発者が非常にインタラクティブなアプリケーションを簡単に作成できるようにするかを実証します。

リアルタイム通信のコアコンセプト

実装の詳細に入る前に、Pythonにおけるリアルタイム通信の基盤となるコアコンセプトを理解することが不可欠です。

WebSockets: WebSocketは、単一の永続的なTCP接続を介してフルデュプレックス通信チャネルを提供します。ステートレスであり、通常は各リクエスト後に接続を閉じるHTTPとは異なり、WebSocket接続は開いたままになり、クライアントとサーバー間の継続的で双方向のデータ交換を可能にします。これは、従来のポーリング手法と比較して、レイテンシとオーバーヘッドを大幅に削減します。

FastAPI: FastAPIは、Python 3.7+でPythonの型ヒントに基づいてAPIを構築するための、モダンで高速（高性能）なWebフレームワークです。Starlette（Web部分用）とPydantic（データ検証とシリアル化用）上に構築されています。FastAPIは本質的に非同期プログラミング（async/await）をサポートしており、多数のWebSocket接続を効率的に処理するための優れた選択肢となります。

非同期プログラミング (async/await): Pythonのasyncioライブラリとasync/await構文は、メインスレッドをブロックすることなく多数の同時WebSocket接続を処理するための基本です。これにより、アプリケーションは多数のクライアントを同時に管理でき、応答性とスケーラビリティを確保できます。

接続管理: リアルタイムアプリケーションでは、サーバーはアクティブなWebSocket接続を管理する必要があります。クライアントが接続すると、そのWebSocketオブジェクトは通常（リストや辞書などに）保存され、サーバーが特定のクライアントにメッセージを送信したり、すべての接続済みのクライアントにブロードキャストしたりできるようにします。同様に、サーバーは非アクティブなクライアントを削除するために切断イベントを処理する必要があります。

リアルタイム通信の実装

FastAPIとWebSocketを使用して基本的なリアルタイム通信システムを実装する方法を探ってみましょう。例として、シンプルなチャットアプリケーションを構築します。

1. FastAPIのセットアップ

まず、FastAPIとUvicorn（ASGIサーバー）がインストールされていることを確認してください。

pip install fastapi uvicorn websockets

2. 基本的なWebSocketエンドポイント

FastAPIで基本的なWebSocketエンドポイントを定義する方法は次のとおりです。

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

app = FastAPI()

# アクティブなWebSocket接続を保存
websocket_connections: List[WebSocket] = []

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()  # WebSocket接続を受け入れる
    websocket_connections.append(websocket)
    try:
        while True:
            data = await websocket.receive_text()  # クライアントからメッセージを受信する
            print(f"Received message: {data}")
            # 送信者にメッセージをエコーバックする
            await websocket.send_text(f"Message text was: {data}")

            # オプション：すべての接続済みのクライアントにブロードキャストする
            # for connection in websocket_connections:
            #     if connection != websocket: # 送信者には返送しない
            #         await connection.send_text(f"Someone said: {data}")
    except WebSocketDisconnect:
        websocket_connections.remove(websocket)
        print("Client disconnected")

# アプリケーションを実行するには： uvicorn main:app --reload

この例では：

  • @app.websocket("/ws")はWebSocketルートを定義します。
  • websocket_endpoint関数はasync関数であり、WebSocketオブジェクトを受け入れます。
  • await websocket.accept()はWebSocket接続を確立します。
  • while Trueループはawait websocket.receive_text()を使用して、着信メッセージを継続的にリッスンします。
  • await websocket.send_text()はクライアントにメッセージを返送します。
  • WebSocketDisconnect例外は、クライアントが切断したときにキャッチされ、websocket_connectionsリストをクリーンアップできるようにします。

3. メッセージのブロードキャスト

チャットアプリケーションの場合、すべての接続済みのクライアントにメッセージをブロードキャストすることは不可欠です。この例を強化して、これをサポートしましょう。

# main.py (続き)
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def send_personal_message(self, message: str, websocket: WebSocket):
        await websocket.send_text(message)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"Client #{client_id} says: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"Client #{client_id} left the chat")

# テスト用にシンプルなHTMLクライアントを提供する
from fastapi.responses import HTMLResponse

@app.get("/")
async def get():
    return HTMLResponse("""
        <!DOCTYPE html>
        <html>
            <head>
                <title>Chat App</title>
            </head>
            <body>
                <h1>WebSocket Chat</h1>
                <form action="" onsubmit="sendMessage(event)">
                    <input type="text" id="messageText" autocomplete="off"/>
                    <button>Send</button>
                </form>
                <ul id='messages'>
                </ul>
                <script>
                    var ws = new WebSocket("ws://localhost:8000/ws/1"); // クライアント1として接続
                    ws.onopen = function(event) {
                        console.log("WebSocket connection opened:", event);
                    };
                    ws.onmessage = function(event) {
                        var messages = document.getElementById('messages')
                        var message = document.createElement('li')
                        var content = document.createTextNode(event.data)
                        message.appendChild(content)
                        messages.appendChild(message)
                    };
                    ws.onclose = function(event) {
                        console.log("WebSocket connection closed:", event);
                    };
                    ws.onerror = function(event) {
                        console.error("WebSocket error observed:", event);
                    };
                    function sendMessage(event) {
                        var input = document.getElementById("messageText")
                        ws.send(input.value)
                        input.value = ''
                        event.preventDefault()
                    }
                </script>
            </body>
        </html>
    """)

この強化されたバージョンでは：

  • アクティブな接続の管理、接続、切断、およびメッセージのブロードキャストのロジックをカプセル化するためにConnectionManagerクラスを導入しました。これにより、コードがクリーンになり、整理が向上します。
  • WebSocketエンドポイントは、クライアントを識別するために使用できるclient_idパスパラメータを取るようになりました。
  • メッセージを受信すると、manager.broadcast()はそれをすべての接続済みのクライアントに送信します。
  • チャットアプリケーションをブラウザで直接テストするために、JavaScriptを備えたシンプルなHTMLページが提供されます。

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

WebSocketのパワーは、数多くのリアルタイムアプリケーションシナリオに及びます。

  • ライブチャットアプリケーション: 実証したように、ユーザー間のリアルタイムメッセージ交換。
  • マルチプレイヤーゲーム: クライアント間のゲーム状態とプレイヤーアクションの同期。
  • 共同編集: 複数のユーザーが同時にドキュメントを編集するGoogleドキュメントのようなアプリケーション。
  • ダッシュボードアップデート: リアルタイムメトリクス、株価、またはセンサーデータをダッシュボードにプッシュ。
  • 通知: ユーザーがリフレッシュする必要なしに、インスタントアラートと通知。
  • IoTデバイス通信: コマンド＆コントロールまたはデータストリーミングのためのIoTデバイスとの双方向通信。

結論

リアルタイム通信は、最新のインタラクティブアプリケーションの基盤であり、Python、特にFastAPIのようなフレームワークは、WebSocketを使用してそれを実装するための堅牢で効率的な方法を提供します。永続的なフルデュプレックス接続を確立することにより、WebSocketはレイテンシを大幅に削減し、従来のポーリングと比較して優れたユーザーエクスペリエンスを提供します。FastAPIの非同期性質と明確なAPI設計により、これらのリアルタイムシステムの構築とスケーリングが容易になり、開発者はダイナミックで応答性の高いアプリケーションを簡単に作成できます。FastAPIでWebSocketをマスターすることは、Pythonで強化されたWebサービスにインタラクティビティの新しい次元を解き放ちます。

Share this article

HackerNews

More Posts from Leapcell

cover

SQLAlchemy CoreでネイティブSQLのパワーを解き放つ

cover

PelicanとMkDocsでコンテンツをスーパーチャージする

Popular Posts