FastAPIのWebSocket：エンドポイント、配信、テスト
FastAPIのWebSocket
FastAPIのWebSocket：エンドポイント、配信マネージャー、テスト | We.Inc
FastAPIでWebSocketエンドポイントを書き、接続マネージャーで全クライアントに配信し、クエリのトークンを読み、切断を処理して、TestClientでまとめてテストする方法。
FastAPIでは、@app.websocket("/ws")と、WebSocketを受け取る非同期関数でソケットのルートを宣言します。await websocket.accept()を呼び出してから、receive_text()とsend_text()をループで回します。クライアントが離脱したときの後始末にはWebSocketDisconnectを捕捉します。最初にwebsocketsパッケージをインストールし、テストにはTestClient.websocket_connect()を使います。
よく呼び出すWebSocketのメソッド
- await websocket.accept()
- ハンドシェイクを完了する。これより前には何も送信できない
- await websocket.receive_text()
- 次のテキストメッセージを待つ
- await websocket.receive_json()
- メッセージを待ち、JSONとして解析する
- await websocket.send_text(s)
- テキストメッセージを送信する
- await websocket.send_json(obj)
- オブジェクトをシリアライズして送信する
- await websocket.close(code=1000)
- クローズコードを付けて接続を終了する
下のエンドポイントビルダーは、いくつかの選択肢から動作するmain.pyを書き出します。ルートのパス、メッセージがプレーンテキストかJSONか、全クライアントが全メッセージを受け取るか、クエリ文字列でトークンを必須にするか、です。対応するpytestファイルも付いてきます。どのオプションもFastAPIのドキュメントにあるパターンに対応しており、ここでは実行可能な1つのファイルにまとめ直しています。
- ビルダーでオプションを選び、main.pyを空のフォルダーにコピーします。
- 必要なものをインストールします：pip install "fastapi[standard]" websockets。その後fastapi dev main.pyでアプリを起動します。
- We.IncのWebSocketテスターかブラウザーのコンソールから、ws://localhost:8000に自分のパスを付けて接続します。
- 生成されたテストファイルをコピーしてpytestを実行すれば、サーバーを起動せずにエンドポイントを確認できます。
- 受信ループはtryとexcept WebSocketDisconnectで囲みましょう。これがないと、タブを閉じたクライアントのたびにスタックトレースがログに残ります。
- accept()の前にソケットを閉じると、サーバーはハンドシェイクにHTTP 403で応答し、クライアントの接続は開きません。読み取れるポリシー違反のコードをクライアントに受け取らせたい場合は、先に受け入れてから1008で閉じましょう。
- メモリ上の接続リストが把握しているのは、現在のプロセスのクライアントだけです。複数のワーカーを動かすと、それぞれが異なる一部のクライアントにだけ配信します。
- メッセージに複数のフィールドが含まれるようになったら、receive_jsonとsend_jsonを使いましょう。json.loadsの呼び出しが不要になり、不正な入力では明確にエラーになります。
- ブラウザーのクライアントはソケットにAuthorizationヘッダーを追加できないため、認証情報はクエリパラメーターかCookieから読み取りましょう。
よくある質問
FastAPIのWebSocketが404を返したり、アップグレードに失敗したりするのはなぜですか？
まずインストールを確認しましょう。FastAPIのWebSocketのドキュメントでは、FastAPIに加えてwebsocketsパッケージを追加するよう求めています。次にパスを確認します。@app.websocketで宣言したルートはソケットのアップグレードにだけ応答し、通常のHTTPのGETには応答しないからです。accept()の前にソケットを閉じた場合も、ハンドシェイクの拒否（HTTP 403）として現れます。
すべてのクライアントにメッセージを配信するには？
受け入れたソケットを小さなマネージャークラス内のリストに保持し、接続時に追加、切断時に削除して、リストをループしながらsend_textを呼び出します。上のビルダーで配信オプションにチェックを入れると、まさにこのコードが生成されます。
複数のUvicornワーカーでも配信は機能しますか？
メモリ上のリストでは機能しません。各ワーカーは独自のリストを持つ別プロセスだからです。FastAPIのドキュメントはencode/broadcasterを紹介しています。これはRedisやPostgreSQLを介してメッセージを中継し、すべてのプロセスがすべてのメッセージを受け取れるようにします。
WebSocketのルートでDependsやQueryを使えますか？
使えます。WebSocketのエンドポイントはHTTPのルートと同じくDepends、Query、Cookie、Header、Path、Securityを受け付けるため、認証の依存関係を両方で共有できます。
WebSocketエンドポイントはどうテストしますか？
アプリのTestClientを作成し、withブロック内でclient.websocket_connect("/ws")を使ってソケットを開きます。その中でsend_textとreceive_textを使ってやり取りを進め、ブロックを抜けると接続が閉じます。
エンドポイントのループが行っていること
- ソケットのハンドラーは長時間動き続けるコルーチンです。一度受け入れた後は、次のメッセージが来るまでawait receive_text()で待機するため、待機中のクライアント1つにかかるコストはほぼゼロです。クライアントがいなくなると、待機中の受信がWebSocketDisconnectを送出します。そこが後始末のタイミングです。
- ハンドラーは非同期なので、あるクライアントが待機している間も他のクライアントへの処理は続きます。ループ内ではtime.sleepや同期型のデータベースドライバーのようなブロッキング呼び出しを避けましょう。ワーカー内のすべての接続が止まってしまいます。
受け入れ前の認証
- トークンはクエリ文字列から通常のパラメーターとして読み取ります。拒否する方法は2つあります。accept()の前に閉じると、ハンドシェイクでHTTP 403になり、ブラウザーは接続の失敗としか報告しません。受け入れてからコード1008で閉じれば、クライアントは理由を読み取れるため、延々とリトライせずにサインインを促す表示を出せます。
動作する最小のエンドポイント
main.py：エコー
受け入れた後、クライアントが切断するまで各メッセージをそのまま返します。
接続マネージャーで配信する
すべての接続で共有するマネージャークラス
単一プロセスで動作します。スケールアウトする前に、リストをRedisベースの中継に置き換えましょう。
TestClientでテストする
サーバーは不要です。クライアントがプロセス内でアプリと直接やり取りします。pytestで実行してください。
FastAPIのWebSocketエンドポイントビルダー
エンドポイントの動作を選び、完全なmain.pyと、それを検証するpytestファイルをコピーしましょう。
メッセージの形式
プレーンテキスト
JSON
すべてのメッセージを接続中の全クライアントに送る
クエリ文字列でトークンを必須にする（?token=...）
実行：fastapi dev main.py の後、ws://localhost:8000{v} に接続
無料で始める · クレジットカード不要
関連: ガイド｜Webマーケティングと集客を学ぶ
製品
ご利用いただける方
機能
リソース
会社情報
サイトマップを見る