FastAPIで極める非同期処理：バックグラウンドタスクとWebSocket活用ガイド
📋 記事のポイント（サマリー）
- 誰向け？
- 同期処理のAPIは書けるが、非同期やリアルタイム通信に挑戦したい中級者〜上級者
- 学べること
async def/awaitを用いた非同期関数の基本- FastAPIのバックグラウンドタスク機能で非同期バッチ処理を実装
- WebSocketエンドポイントの立ち上げとクライアント連携
- フロントエンドとの双方向リアルタイム通信サンプル
- 実運用時の注意点（接続数制限・例外処理など）
- 得られる効果
- 高負荷／長時間処理をAPIレスポンスと切り分け、ユーザー体験向上
- チャットや通知などリアルタイム機能を簡単に組み込める
🎯 対象読者
- 開発者Dさん（27歳・社会人）
すでにREST APIは同期版で書けるが、画像処理やメール送信を非同期に回したい方 - エンジニアEさん（30代）
チャット機能を学習中で、FastAPIでWebSocketを動かしてみたい方 - サイドプロジェクトFさん（24歳・学生）
サーバー負荷を抑えつつリアルタイムゲームのプロトタイプを作りたい方
♿ アクセシビリティレベル
- 読みやすさ：短めの文と箇条書きで視線移動を減らし、重要語はルビで補足
- 構造化：見出し・サブ見出しを階層化し、スクリーンリーダーで論理的に把握可能
- コード例：固定幅フォントで改行やインデントを揃え、コメントを充実
- 要約：各章末にポイントをまとめ、後から見返しやすい
1. 非同期処理の基本：async/awaitとは？
Python 3.5以降で導入された 非同期関数 は、async def／await 構文で記述します。
- 同期処理：関数が終わるまで他の処理を待つ
- 非同期処理：重たい処理中も別タスクを並行実行できる
import asyncio
async def say_after(delay: int, message: str):
await asyncio.sleep(delay)
print(message)
async def main():
# 並行実行（ここではタスクをまとめて実行）
await asyncio.gather(
say_after(1, "こんにちは"),
say_after(2, "FastAPI！"),
)
# 実行
asyncio.run(main())
要点まとめ
async defで非同期関数を定義awaitで他の非同期処理が終わるのを待機asyncio.gatherで複数タスクを同時実行
2. バックグラウンドタスクで長時間処理を切り分け
2.1 FastAPIの BackgroundTasks
FastAPI標準の BackgroundTasks を使うと、レスポンス返却後に別処理を実行できます。
from fastapi import FastAPI, BackgroundTasks
import time
app = FastAPI()
def write_log(message: str):
time.sleep(5) # 重たい処理（例：ファイルI/O）
with open("log.txt", "a") as f:
f.write(message + "\n")
@app.post("/process/")
async def process_data(data: dict, background_tasks: BackgroundTasks):
# レスポンスを返した後にバックグラウンドで write_log が動く
background_tasks.add_task(write_log, f"Processed: {data}")
return {"status": "accepted"}
- リクエスト→即レスポンス→裏で長時間処理
- ユーザーは待たずに画面を更新可能
要点まとめ
- エンドポイント引数に
BackgroundTasksを追加add_taskで同期／非同期関数両方を処理後にキューイング
2.2 複数タスク・例外処理
async def notify_user(user_id: int, message: str):
# 非同期関数もキューイング可
await asyncio.sleep(2)
# ここでメール送信やSNS通知を実装
@app.post("/notify/{user_id}")
async def notify(user_id: int, background_tasks: BackgroundTasks):
background_tasks.add_task(notify_user, user_id, "お知らせです")
return {"detail": "通知中"}
notify_userはasync defなので非同期で並行実行- タスク内で例外が起きてもAPI本体には影響なし
3. WebSocketでリアルタイム通信を実装
3.1 WebSocketの基本エンドポイント
FastAPIでは websockets 標準を利用し、簡単にWebSocketを立ち上げられます。
from fastapi import FastAPI, WebSocket
app = FastAPI()
@app.websocket("/ws")
async def websocket_endpoint(ws: WebSocket):
await ws.accept()
while True:
data = await ws.receive_text()
# 受信内容をそのまま返信
await ws.send_text(f"あなた: {data}")
- クライアントから送られたテキストを即座に返す エコーサーバ
while Trueで接続維持しつつ、双方向通信
要点まとめ
@app.websocketでエンドポイント定義accept()→receive_XXX()→send_XXX()
3.2 フロントエンドから接続する例
<!DOCTYPE html>
<html>
<head><meta charset="utf-8"><title>FastAPI WS</title></head>
<body>
<input id="msg" placeholder="メッセージ入力">
<button onclick="send()">送信</button>
<pre id="log"></pre>
<script>
const ws = new WebSocket("ws://127.0.0.1:8000/ws");
ws.onmessage = e => {
document.getElementById("log").textContent += e.data + "\n";
};
function send() {
const input = document.getElementById("msg");
ws.send(input.value);
input.value = "";
}
</script>
</body>
</html>
- ブラウザで開くだけでリアルタイムチャット体験
- CORS設定不要で同一オリジンならそのまま動作
4. 実運用での注意点
- 接続数制限
- WebSocketは常に接続維持 → サーバリソースを圧迫します
- Nginx／uvicornの
--wsオプションで同時接続数を制御
- 例外／切断処理
- クライアントが突然切断すると例外が発生 →
try/except WebSocketDisconnectでハンドリング
- クライアントが突然切断すると例外が発生 →
- バックグラウンドタスクの監視
- 大量タスク時は Celery などの専用キューを導入検討
- セキュリティ
- WebSocket経由の認証はトークンをクエリやヘッダで渡す方法を採用
要点まとめ
- 負荷対策と例外制御を組み込んで安定稼働を目指す
- 必要に応じて専用キュー／メッセージブローカーを活用
5. まとめと次のステップ
本記事では、FastAPIを使った非同期処理の２大機能、バックグラウンドタスク と WebSocket を中心に解説しました。
async/awaitの基礎BackgroundTasksでレスポンス即返却＋長時間処理@app.websocketでリアルタイム通信- フロント連携サンプルHTML
- 実運用での制限・例外・セキュリティ注意点
次は、以下を試してみましょう：
- メッセージブローカー連携：RedisやRabbitMQを使ったPub/Subモデル
- フロントフレームワーク統合：React/Vueでさらに洗練されたチャットUI
- 分散タスクキュー：CeleryやFastAPI-Backgroundタスクでスケーラブルに
FastAPIの非同期機能は拡張性が高く、用途に合わせて多彩な実装が可能です。ぜひ実践しながら、自分だけのリアルタイムAPIを作り込んでみてくださいね♡