Lesson 16 · FastAPI deep dive

FastAPI Tutorial #16: Real-Time WebSockets

Video sixteen of the eighteen-part series: a real persistent, two-way connection instead of request-response. Accepting connections, sending and receiving,…

⬇ Download notebook

CourseFastAPI deep dive
Lesson16 of 18
FormatJupyter notebook · 10 code cells

FastAPI

What you'll learn

The Problem - HTTP Is Genuinely One-Way, Per Request
@app.websocket - a Persistent Connection
A Loop - Handling Many Messages on One Connection
JSON Messages - receivejson and sendjson
WebSocketDisconnect - Handling a Client Leaving
A ConnectionManager - Tracking Multiple Clients
broadcast - Sending to Every Connected Client
Two Clients, One Broadcast

Data

No separate download needed — the notebook creates or downloads everything it uses.

📓 Full notebook

Download .ipynb

FastAPI Deep-Dive, Video 16: WebSockets#

Video sixteen of the eighteen-part series: a real persistent, two-way connection instead of request-response.
Accepting connections, sending and receiving, disconnect handling, and broadcasting to many clients.
Let's get into it.

Part 1: The Problem - HTTP Is Genuinely One-Way, Per Request#

from fastapi import FastAPI
from fastapi.testclient import TestClient
app = FastAPI()
@app.get('/price')
def get_price():
    return {'price': 9.99}
client = TestClient(app)
print(client.get('/price').json())

{'price': 9.99}

Part 2: @app.websocket - a Persistent Connection#

from fastapi import WebSocket
@app.websocket('/ws')
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    data = await websocket.receive_text()
    await websocket.send_text(f'echo: {data}')
with client.websocket_connect('/ws') as ws:
    ws.send_text('hello')
    print(ws.receive_text())

echo: hello

Part 3: A Loop - Handling Many Messages on One Connection#

from fastapi import WebSocketDisconnect
@app.websocket('/counter')
async def counter_endpoint(websocket: WebSocket):
    await websocket.accept()
    count = 0
    try:
        while True:
            await websocket.receive_text()
            count += 1
            await websocket.send_text(str(count))
    except WebSocketDisconnect:
        pass
with client.websocket_connect('/counter') as ws:
    ws.send_text('ping')
    print(ws.receive_text())
    ws.send_text('ping')
    print(ws.receive_text())

1
2

Part 4: JSON Messages - receive_json and send_json#

@app.websocket('/echo-json')
async def echo_json_endpoint(websocket: WebSocket):
    await websocket.accept()
    payload = await websocket.receive_json()
    await websocket.send_json({'received': payload})
with client.websocket_connect('/echo-json') as ws:
    ws.send_json({'item': 'mug', 'qty': 2})
    print(ws.receive_json())

{'received': {'item': 'mug', 'qty': 2}}

Part 5: WebSocketDisconnect - Handling a Client Leaving#

from fastapi import WebSocketDisconnect
disconnect_log = []
@app.websocket('/tracked')
async def tracked_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            msg = await websocket.receive_text()
            await websocket.send_text(f'got: {msg}')
    except WebSocketDisconnect:
        disconnect_log.append('client left')
with client.websocket_connect('/tracked') as ws:
    ws.send_text('hi')
    print(ws.receive_text())
print(disconnect_log)

got: hi
['client left']

Part 6: A ConnectionManager - Tracking Multiple Clients#

class ConnectionManager:
    def __init__(self):
        self.active = []
    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active.append(websocket)
    def disconnect(self, websocket: WebSocket):
        self.active.remove(websocket)
manager = ConnectionManager()
print(len(manager.active))

Part 7: broadcast - Sending to Every Connected Client#

async def broadcast(manager: ConnectionManager, message: str):
    for connection in manager.active:
        await connection.send_text(message)
@app.websocket('/chat')
async def chat_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            msg = await websocket.receive_text()
            await broadcast(manager, f'someone said: {msg}')
    except WebSocketDisconnect:
        manager.disconnect(websocket)

Part 8: Two Clients, One Broadcast#

with client.websocket_connect('/chat') as ws1, client.websocket_connect('/chat') as ws2:
    ws1.send_text('hi everyone')
    print(ws1.receive_text())
    print(ws2.receive_text())

someone said: hi everyone
someone said: hi everyone

Part 9: Path Parameters on a WebSocket Route#

@app.websocket('/rooms/{room_name}')
async def room_endpoint(websocket: WebSocket, room_name: str):
    await websocket.accept()
    await websocket.send_text(f'joined room: {room_name}')
with client.websocket_connect('/rooms/python-fans') as ws:
    print(ws.receive_text())

joined room: python-fans

Part 10: A Real Pattern - a Small Live Chat Room#

chat_app = FastAPI()
class ChatManager:
    def __init__(self):
        self.connections = {}
    async def join(self, websocket: WebSocket, username: str):
        await websocket.accept()
        self.connections[websocket] = username
    def leave(self, websocket: WebSocket):
        self.connections.pop(websocket, None)
    async def broadcast(self, message: str):
        for connection in list(self.connections.keys()):
            await connection.send_text(message)
chat_manager = ChatManager()
@chat_app.websocket('/rooms/{room_name}/{username}')
async def chat_room(websocket: WebSocket, room_name: str, username: str):
    await chat_manager.join(websocket, username)
    await chat_manager.broadcast(f'{username} joined {room_name}')
    try:
        while True:
            msg = await websocket.receive_text()
            await chat_manager.broadcast(f'{username}: {msg}')
    except WebSocketDisconnect:
        chat_manager.leave(websocket)
        await chat_manager.broadcast(f'{username} left {room_name}')
chat_client = TestClient(chat_app)
with chat_client.websocket_connect('/rooms/python/alex') as ws1:
    print(ws1.receive_text())
    with chat_client.websocket_connect('/rooms/python/sam') as ws2:
        print(ws1.receive_text())
        print(ws2.receive_text())
        ws2.send_text('hello alex')
        print(ws1.receive_text())

alex joined python
sam joined python
sam joined python
sam: hello alex

Wrap-Up: What You Learned#

HTTP is one request, one response; WebSockets keep a persistent, two-way connection open.
@app.websocket declares a websocket route; websocket.accept() upgrades and opens the connection.
A while True loop lets one connection handle as many messages as the client sends.
receive_json and send_json exchange structured data, handling serialization automatically.
WebSocketDisconnect is raised when the client leaves, letting the server clean up state.
A ConnectionManager class centralizes tracking every currently connected client.
A broadcast method loops over tracked connections, sending the same message to each.
Two clients connected to the same route can both receive a message sent by either one.
A websocket route accepts path parameters exactly like a normal HTTP endpoint does.
That wraps up WebSockets. Next up: Testing FastAPI Apps.

Found this useful?

All lessons, notebooks and datasets here are free. If they helped you, a coffee keeps new lessons coming.

☕ Buy me a coffee Other ways to help