第3部:AIとバックエンド高度化 Step 11 / 36

API設計の高度化

GraphQL、WebSocket、Server-Sent Events など、高度なAPI技術をAIと一緒に実装します。

WebSocket によるリアルタイム通信

AIへの指示

FastAPI で WebSocket を使ったリアルタイムチャット機能を実装してください。

【要件】
- ルームごとにメッセージを配信
- 接続・切断の管理
- 認証済みユーザーのみ接続可能

WebSocket 実装例

from fastapi import WebSocket, WebSocketDisconnect

class ConnectionManager:
    def __init__(self):
        self.active_connections: dict[str, list[WebSocket]] = {}

    async def connect(self, websocket: WebSocket, room_id: str):
        await websocket.accept()
        if room_id not in self.active_connections:
            self.active_connections[room_id] = []
        self.active_connections[room_id].append(websocket)

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

manager = ConnectionManager()

@app.websocket("/ws/{room_id}")
async def websocket_endpoint(websocket: WebSocket, room_id: str):
    await manager.connect(websocket, room_id)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(room_id, data)
    except WebSocketDisconnect:
        manager.disconnect(websocket, room_id)

GraphQL の導入

REST vs GraphQL

REST

  • ・エンドポイントごとに固定レスポンス
  • ・オーバーフェッチ/アンダーフェッチ
  • ・シンプルで理解しやすい

GraphQL

  • ・クライアントが必要なデータを指定
  • ・1リクエストで複数リソース取得
  • ・学習コストがある

AIへの指示

Strawberry (Python GraphQL) を使って、
既存のFastAPI RESTエンドポイントをGraphQLに変換してください。

User と Post の関連を持つスキーマを定義してください。

まとめ

  • WebSocket - 双方向リアルタイム通信に最適
  • SSE - サーバーからの一方向プッシュに
  • GraphQL - 柔軟なデータ取得が必要な場合に
データベース最適化 次へ:パフォーマンス改善