FastAPI Backend Development Bootcamp · レッスン

リアルタイム通信のためのWebSockets

WebSocketsを使ってFastAPIにリアルタイム機能を構築します。接続の受け付け、メッセージの交換、多数のクライアントへのブロードキャスト、asyncパターンによる切断処理を学びます。

レッスン 4/413 ステップ

「リアルタイム通信のためのWebSockets」はCoddyKit上の無料FastAPI Backend Development Bootcampレッスンです。 これはレッスン4/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはFastAPI Backend Development Bootcamp学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 FastAPI Backend Development Bootcampコースには全4レッスンが含まれています。

このレッスンの一部はまだ翻訳されておらず、英語で表示されています。

From Request/Response to Real-Time

Regular HTTP is request/response: the client asks, the server answers, the connection closes. WebSockets keep a single connection open for full-duplex, real-time messaging. Perfect for chat, live dashboards, and notifications.

Why Async Fits WebSockets

A WebSocket connection lives a long time and is mostly idle waiting for messages. FastAPI's async model lets one worker handle thousands of concurrent connections without blocking.

A Minimal WebSocket Endpoint

Declare a @app.websocket route. Accept the connection, then loop receiving and sending messages.

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket('/ws')
async def ws(websocket: WebSocket):
    await websocket.accept()
    while True:
        msg = await websocket.receive_text()
        await websocket.send_text(f'echo: {msg}')

Sending JSON

You can exchange structured data with receive_json and send_json instead of raw text.

data = await websocket.receive_json()
await websocket.send_json({'received': data, 'status': 'ok'})

Handling Disconnects

Clients drop off. Catch WebSocketDisconnect to clean up resources instead of crashing the handler.

from fastapi import WebSocketDisconnect

try:
    while True:
        await websocket.receive_text()
except WebSocketDisconnect:
    print('client left')

A Connection Manager

To broadcast, track active connections in a manager class. Add on connect, remove on disconnect.

class Manager:
    def __init__(self):
        self.active = []
    async def connect(self, ws):
        await ws.accept()
        self.active.append(ws)
    def disconnect(self, ws):
        self.active.remove(ws)

Broadcasting to Everyone

Loop the active connections and send each one the message. This is the core of a chat room.

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

Broadcast Logic in Plain Python

The manager is just list bookkeeping. Here is the connect/disconnect/broadcast flow simulated synchronously.

active = []
def connect(c): active.append(c)
def disconnect(c): active.remove(c)
def broadcast(msg): return [f'{c}<-{msg}' for c in active]
connect('alice'); connect('bob')
print(broadcast('hi'))
disconnect('alice')
print(broadcast('bye'))

A Chat Room Endpoint

Combine the pieces: connect, broadcast each received message, and disconnect on drop.

manager = Manager()

@app.websocket('/chat')
async def chat(ws: WebSocket):
    await manager.connect(ws)
    try:
        while True:
            text = await ws.receive_text()
            await manager.broadcast(text)
    except WebSocketDisconnect:
        manager.disconnect(ws)

Scaling Beyond One Process

An in-memory manager only knows connections on its worker. To broadcast across multiple workers or servers, use a pub/sub backend like Redis to fan out messages.

Security and Limits

Protect WebSocket endpoints:

  • Authenticate during the handshake (e.g. token in query or header).
  • Validate and size-limit incoming messages.
  • Heartbeat/ping to detect dead connections.

Quick Check

Why does an in-memory connection manager fail to broadcast correctly when you run several Uvicorn workers?

Recap

You added real-time communication:

  • Accepted WebSocket connections and exchanged text/JSON.
  • Handled WebSocketDisconnect cleanly.
  • Built a connection manager to broadcast to many clients.
  • Learned to scale with Redis pub/sub and to secure connections.
無料で開始

AI チューターと学ぶ FastAPI Backend Development Bootcamp — 無料

ブラウザでリアルコードを書いて実行し、24/7 の AI チューターから瞬時にサポートを受け、ウェブまたはアプリで続きから学習できます。

コース
21
レッスン
84

よくある質問

「リアルタイム通信のためのWebSockets」レッスンは無料ですか?

はい。「リアルタイム通信のためのWebSockets」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し(組み込みコードエディタと24時間対応のAIチューター)、FastAPI Backend Development Bootcampコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 FastAPI Backend Development Bootcampコースには全4レッスンが含まれています。

「リアルタイム通信のためのWebSockets」で何を学びますか?

WebSocketsを使ってFastAPIにリアルタイム機能を構築します。接続の受け付け、メッセージの交換、多数のクライアントへのブロードキャスト、asyncパターンによる切断処理を学びます。 ブラウザで直接実行するハンズオンコードでFastAPI Backend Development Bootcampを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。

FastAPI Backend Development Bootcampを始めるのに経験は必要ですか?

事前経験は必要ありません。CoddyKitのFastAPI Backend Development Bootcampは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン4/4です。

「リアルタイム通信のためのWebSockets」レッスンにはどのくらい時間がかかりますか?

ほとんどのCoddyKitレッスンは約5~10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。

このFastAPI Backend Development Bootcampレッスンでコードを書いて実行できますか?

はい。すべてのFastAPI Backend Development Bootcampレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。

このコースのすべてのレッスン

  1. PythonのAsync/Await入門
  2. FastAPIと非同期処理
  3. バックグラウンドタスクの実行
  4. リアルタイム通信のためのWebSockets
← FastAPI Backend Development Bootcampに戻る