WebSocket — Real-Time Bidirectional Communication
HTTP is like sending a letter—it goes one way and comes back; WebSocket is like making a phone call—both parties can talk at any time without having to hang up and redial.
1. What You'll Learn
- WebSocket Basics:
@app.websocketDecorators and Lifecycle (connect/receive/disconnect) - Connection Manager Pattern:
ConnectionManagerClass Design and Broadcasting Mechanism - Authenticated WebSocket: Verify the JWT token during the handshake phase
- Integrates with HTTP endpoints: Price changes trigger WebSocket push notifications
- Alice Scenario: Real-time Price Feed—When the price of a product changes among millions of items, subscribers immediately receive a push notification
2. Alice's True Story
(1) Pain Point: Price changes can only be retrieved through polling
Bob's frontend polls the PriceTracker API every 5 seconds to check for price changes, but 99% of the prices for millions of products remain unchanged within any given 5-second period, meaning 99% of the requests are wasted. To make matters worse, it can take up to 5 seconds for price changes to be displayed, leading to customer complaints that "prices aren't real-time."
(2) WebSocket Solution
WebSocket establishes a persistent, bidirectional connection and actively pushes price changes from the server to the front end when they occur, eliminating the need for polling—zero waste, zero latency.
PYTHON
@app.websocket("/ws/prices")
async def price_websocket(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_json({"price_update": data})
(3) Revenue
The number of polling requests dropped from 200 per second to 0, the price update delay decreased from 5 seconds to 50 ms, Bob's front end no longer wastes API quotas, and customer satisfaction has improved significantly.
3. WebSocket Basics
(1) Lifecycle States
stateDiagram-v2
[*] --> CONNECTING: Client initiates
CONNECTING --> CONNECTED: accept()
CONNECTED --> RECEIVING: receive()
RECEIVING --> CONNECTED: send()
CONNECTED --> CLOSING: close() / disconnect
CLOSING --> CLOSED: Connection closed
CLOSED --> [*]
(1) ▶ Example: A Minimal WebSocket Endpoint
PYTHON
from fastapi import FastAPI, WebSocket
app = FastAPI()
@app.websocket("/ws/echo")
async def websocket_echo(websocket: WebSocket):
await websocket.accept() # Accept connection
try:
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Echo: {data}")
except Exception:
await websocket.close()
Output:
TEXT
# Function defined successfully
(2) ▶ Example: WebSocket with Path Parameters
PYTHON
@app.websocket("/ws/products/{product_id}/prices")
async def product_price_stream(
websocket: WebSocket,
product_id: int,
):
await websocket.accept()
try:
while True:
data = await websocket.receive_json()
# Echo back with product context
await websocket.send_json({
"product_id": product_id,
"price": data.get("price"),
"currency": data.get("currency", "USD"),
})
except Exception:
await websocket.close()
Output:
TEXT
# Function defined successfully
4. Connection Manager Mode
(1) ConnectionManager Design
(1) ▶ Example: ConnectionManager Implementation
PYTHON
from fastapi import FastAPI, WebSocket
from typing import Dict, List
import json
class ConnectionManager:
def __init__(self):
# Map: product_id -> list of active connections
self.active_connections: Dict[int, List[WebSocket]] = {}
async def connect(self, websocket: WebSocket, product_id: int):
await websocket.accept()
if product_id not in self.active_connections:
self.active_connections[product_id] = []
self.active_connections[product_id].append(websocket)
def disconnect(self, websocket: WebSocket, product_id: int):
if product_id in self.active_connections:
self.active_connections[product_id].remove(websocket)
if not self.active_connections[product_id]:
del self.active_connections[product_id]
async def broadcast_to_product(self, product_id: int, message: dict):
if product_id in self.active_connections:
dead_connections = []
for connection in self.active_connections[product_id]:
try:
await connection.send_json(message)
except Exception:
dead_connections.append(connection)
# Clean up dead connections
for conn in dead_connections:
self.disconnect(conn, product_id)
async def broadcast_all(self, message: dict):
for product_id in list(self.active_connections.keys()):
await self.broadcast_to_product(product_id, message)
manager = ConnectionManager()
Output:
TEXT
# Function defined successfully
(2) WebSocket Push Architecture
flowchart TD
Update[Price Update via HTTP] --> Handler[API Handler]
Handler --> Manager[ConnectionManager]
Manager --> WS1[WebSocket Client 1]
Manager --> WS2[WebSocket Client 2]
Manager --> WSN[WebSocket Client N]
subgraph Subscribers
WS1
WS2
WSN
end
(2) ▶ Example: Using ConnectionManager with a WebSocket Endpoint
PYTHON
app = FastAPI()
@app.websocket("/ws/products/{product_id}/prices")
async def price_websocket(websocket: WebSocket, product_id: int):
await manager.connect(websocket, product_id)
try:
while True:
# Keep connection alive, receive any client messages
data = await websocket.receive_text()
except Exception:
manager.disconnect(websocket, product_id)
Output:
TEXT
# Function defined successfully
5. Authenticating WebSocket
(1) Verifying the JWT During the Handshake Phase
WebSocket does not have a standard header mechanism; tokens are passed via query parameters.
(1) ▶ Example: WebSocket JWT Authentication
PYTHON
from fastapi import WebSocket, Query, HTTPException
from jose import jwt, JWTError
from app.core.config import settings
async def verify_ws_token(token: str) -> dict:
try:
payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm])
return payload
except JWTError:
raise ValueError("Invalid token")
@app.websocket("/ws/prices")
async def authenticated_price_ws(
websocket: WebSocket,
token: str = Query(..., description="JWT access token"),
):
# Verify token before accepting connection
try:
user = await verify_ws_token(token)
except ValueError:
await websocket.close(code=4001, reason="Authentication failed")
return
await websocket.accept()
try:
while True:
data = await websocket.receive_text()
await websocket.send_json({"user": user.get("sub"), "data": data})
except Exception:
pass
Output:
TEXT
# Function defined successfully
| Authentication Method | Implementation | Advantages | Disadvantages |
|---|---|---|---|
| Query Parameter | ?token=xxx |
Simple | Token appears in URL logs |
| First message | Send token after connecting | Not exposed in the URL | One additional round trip |
| Sec-WebSocket-Protocol | Token transmitted via subprotocol | Not exposed in the URL | Non-standard usage |
6. HTTP and WebSocket Collaboration
(1) Price Changes Trigger Push Notifications
(1) ▶ Example: HTTP Endpoint Triggers a WebSocket Broadcast
PYTHON
from pydantic import BaseModel, Field
from datetime import datetime
class PriceUpdate(BaseModel):
product_id: int = Field(gt=0)
price: float = Field(gt=0, description="New price in USD")
currency: str = Field(default="USD")
source: str = Field(max_length=100)
app = FastAPI()
manager = ConnectionManager()
@app.post("/api/v1/prices", response_model=PriceResponse)
async def create_price(
price: PriceCreate,
db: AsyncSession = Depends(get_db),
user=Depends(get_current_user),
):
service = PriceService(db)
result = await service.create_price(price)
# Trigger WebSocket broadcast after successful price creation
await manager.broadcast_to_product(
product_id=price.product_id,
message={
"event": "price_update",
"product_id": price.product_id,
"new_price": price.price,
"currency": price.currency,
"source": price.source,
"timestamp": datetime.utcnow().isoformat(),
},
)
return result
Output:
TEXT
# Function defined successfully
(2) ▶ Example: Global Price Feed (Subscribe to All Changes)
PYTHON
@app.websocket("/ws/prices/stream")
async def global_price_stream(websocket: WebSocket):
await websocket.accept()
# Add to a global subscriber list
manager.global_connections.append(websocket)
try:
while True:
# Receive heartbeat/ping from client
data = await websocket.receive_text()
if data == "ping":
await websocket.send_text("pong")
except Exception:
manager.global_connections.remove(websocket)
Output:
TEXT
# Function defined successfully
❓ FAQ
Q How do I choose between WebSocket and SSE (Server-Sent Events)?
A Use WebSocket for bidirectional communication (e.g., chat, real-time collaboration), and use SSE when only server-side push is required (e.g., log streams, notifications). SSE is simpler and automatically reconnects.
Q Is there a limit to the number of WebSocket connections?
A On a single machine, the number is limited by file descriptors (typically 65,535). In production environments, Nginx is used for load balancing, and multiple instances share connection state (via Redis Pub/Sub).
Q How should WebSocket disconnections be handled?
A The client implements automatic reconnection (exponential backoff). The server uses ping/pong heartbeats to detect dead connections, and the ConnectionManager automatically cleans them up.
Q Can WebSocket use Pydantic to validate messages?
A Yes. After receiving a message, validate it using
MessageModel.model_validate(data); if validation fails, send an error message to the client.Q How do multiple worker processes broadcast data?
A A single-process, in-memory ConnectionManager isn't sufficient. In production environments, use Redis Pub/Sub: price changes are published to a Redis channel, and each worker subscribes to it and pushes the data to its local connections.
Q How do you handle CORS for WebSockets?
A WebSockets in browsers are also subject to the same-origin policy. FastAPI's CORSMiddleware automatically handles the cross-origin handshake for WebSockets.
📖 Summary
- WebSocket provides a persistent, bidirectional connection, replacing polling to enable true real-time push notifications
- ConnectionManager manages the connection pool and supports broadcasting grouped by product ID
- JWT authentication is verified during the handshake phase using query parameters; if it fails, the connection is closed immediately (code 4001).
- After creating a price for an HTTP endpoint, call
manager.broadcast_to_product()to trigger a push notification - The production environment requires Redis Pub/Sub to enable cross-process broadcasting; single-process memory is not suitable for multiple workers.
📝 Exercises
- Basic Exercise (Difficulty: ⭐): Create a WebSocket echo endpoint where the client sends text and the server returns it exactly as received. Test it using your browser's developer tools or wscat. Hint:
@app.websocket("/ws/echo")+accept()+receive_text() - Advanced Problem (Difficulty ⭐⭐): Implement a ConnectionManager that supports multiple clients subscribing to price changes for the same product. When a new price is created, broadcast it to all WebSocket clients subscribed to that product. Hint:
Dict[int, List[WebSocket]]+broadcast_to_product() - Challenge (Difficulty: ⭐⭐⭐): Add JWT authentication to WebSocket (by passing a token in the query parameters), and trigger a WebSocket broadcast in the HTTP price creation endpoint to implement a complete "HTTP write → WebSocket push" workflow. Hint:
token: str = Query(...)+verify_ws_token()+ HTTP endpoint call tomanager.broadcast_to_product()
---|
