Prerequisites
- Basic understanding of programming concepts 📝
- Python installation (3.8+) 🐍
- VS Code or preferred IDE 💻
What you'll learn
- Understand the concept fundamentals 🎯
- Apply the concept in real projects 🏗️
- Debug common issues 🐛
- Write clean, Pythonic code ✨
🎯 Introduction
Welcome to this exciting tutorial on FastAPI WebSockets! 🎉 In this guide, we’ll explore the magical world of real-time communication between your Python server and clients.
Imagine building a live chat app 💬, a real-time dashboard 📊, or a multiplayer game 🎮 - that’s the power of WebSockets! Unlike traditional HTTP requests that are like sending letters 📮, WebSockets create a phone line 📞 that stays open for instant, two-way communication.
By the end of this tutorial, you’ll be creating amazing real-time features that will make your applications come alive! Let’s dive in! 🏊♂️
📚 Understanding WebSockets
🤔 What are WebSockets?
WebSockets are like having a direct phone line between your server and client! 📱 Think of regular HTTP as sending text messages - you send one, wait for a reply, then send another. WebSockets? That’s a live phone call where both sides can talk whenever they want!
In Python terms, WebSockets provide a persistent, bidirectional communication channel between client and server. This means you can:
- ✨ Send messages instantly without polling
- 🚀 Push updates from server to client immediately
- 🛡️ Maintain a single connection for continuous communication
- 💡 Build truly interactive applications
💡 Why Use WebSockets with FastAPI?
Here’s why developers love WebSockets in FastAPI:
- Real-time Updates 🔄: Push live data without client requests
- Low Latency ⚡: Instant message delivery
- Efficient Resource Usage 💰: One connection, many messages
- Native FastAPI Support 🎯: Built-in WebSocket handling
Real-world example: Imagine building a stock trading app 📈. With WebSockets, you can push price updates instantly to all connected traders - no refresh button needed!
🔧 Basic Syntax and Usage
📝 Simple WebSocket Example
Let’s start with a friendly example:
# 👋 Hello, WebSockets!
from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
app = FastAPI()
# 🎨 Simple HTML client for testing
html = """
<!DOCTYPE html>
<html>
<head>
<title>Chat 💬</title>
</head>
<body>
<h1>WebSocket Chat 🚀</h1>
<form action="" onsubmit="sendMessage(event)">
<input type="text" id="messageText" autocomplete="off"/>
<button>Send 📤</button>
</form>
<ul id='messages'></ul>
<script>
// 🌐 Create WebSocket connection
const ws = new WebSocket("ws://localhost:8000/ws");
// 📨 Handle incoming messages
ws.onmessage = function(event) {
const messages = document.getElementById('messages');
const message = document.createElement('li');
message.textContent = event.data;
messages.appendChild(message);
};
// 📤 Send message function
function sendMessage(event) {
const input = document.getElementById("messageText");
ws.send(input.value);
input.value = '';
event.preventDefault();
}
</script>
</body>
</html>
"""
@app.get("/")
async def get():
return HTMLResponse(html)
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
# 🤝 Accept the connection
await websocket.accept()
# 💬 Echo messages back
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Echo: {data} 🔊")💡 Explanation: This creates a simple echo server! The client connects, sends messages, and receives them back with an echo prefix. Perfect for testing!
🎯 WebSocket Connection Lifecycle
Here’s how WebSocket connections work:
# 🏗️ Complete WebSocket lifecycle
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import asyncio
app = FastAPI()
@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
# 1️⃣ Connection phase
await websocket.accept()
print(f"✅ Client {client_id} connected!")
try:
# 2️⃣ Communication phase
while True:
# 📥 Receive message
message = await websocket.receive_text()
print(f"📨 From {client_id}: {message}")
# 📤 Send response
response = f"Server received: {message} ✨"
await websocket.send_text(response)
except WebSocketDisconnect:
# 3️⃣ Disconnection phase
print(f"👋 Client {client_id} disconnected")💡 Practical Examples
🎮 Example 1: Live Chat Room
Let’s build a real chat application:
# 💬 Real-time chat room
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import Dict, List
import json
from datetime import datetime
app = FastAPI()
# 🏠 Chat room manager
class ConnectionManager:
def __init__(self):
# 👥 Active connections
self.active_connections: Dict[str, WebSocket] = {}
async def connect(self, websocket: WebSocket, user_id: str):
await websocket.accept()
self.active_connections[user_id] = websocket
# 🎉 Announce new user
await self.broadcast(f"{user_id} joined the chat! 👋", "system")
def disconnect(self, user_id: str):
if user_id in self.active_connections:
del self.active_connections[user_id]
async def send_personal_message(self, message: str, websocket: WebSocket):
await websocket.send_text(message)
async def broadcast(self, message: str, sender: str):
# 📢 Send to all connected users
timestamp = datetime.now().strftime("%H:%M:%S")
formatted_message = {
"sender": sender,
"message": message,
"timestamp": timestamp,
"emoji": "💬" if sender != "system" else "📢"
}
disconnected = []
for user_id, connection in self.active_connections.items():
try:
await connection.send_text(json.dumps(formatted_message))
except:
disconnected.append(user_id)
# 🧹 Clean up disconnected users
for user_id in disconnected:
self.disconnect(user_id)
# 🎯 Create manager instance
manager = ConnectionManager()
@app.websocket("/ws/{user_id}")
async def websocket_endpoint(websocket: WebSocket, user_id: str):
await manager.connect(websocket, user_id)
try:
while True:
# 📨 Receive and broadcast messages
data = await websocket.receive_text()
await manager.broadcast(data, user_id)
except WebSocketDisconnect:
manager.disconnect(user_id)
await manager.broadcast(f"{user_id} left the chat 👋", "system")🎯 Try it yourself: Add private messaging between users or typing indicators!
📊 Example 2: Real-time Dashboard
Let’s create a live dashboard:
# 📊 Real-time metrics dashboard
from fastapi import FastAPI, WebSocket
import asyncio
import random
import json
from datetime import datetime
app = FastAPI()
# 📈 Simulated metrics generator
async def generate_metrics():
while True:
metrics = {
"timestamp": datetime.now().isoformat(),
"cpu": random.randint(20, 80),
"memory": random.randint(30, 90),
"requests_per_second": random.randint(100, 1000),
"active_users": random.randint(50, 500),
"emoji_mood": random.choice(["😊", "🚀", "💪", "🎯"])
}
return metrics
# 🎯 Connected dashboards
class DashboardManager:
def __init__(self):
self.connections: List[WebSocket] = []
self.is_broadcasting = False
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.connections.append(websocket)
# 🚀 Start broadcasting if not already running
if not self.is_broadcasting:
self.is_broadcasting = True
asyncio.create_task(self.broadcast_metrics())
def disconnect(self, websocket: WebSocket):
self.connections.remove(websocket)
if not self.connections:
self.is_broadcasting = False
async def broadcast_metrics(self):
while self.is_broadcasting and self.connections:
# 📊 Generate new metrics
metrics = await generate_metrics()
# 📤 Send to all dashboards
disconnected = []
for connection in self.connections:
try:
await connection.send_text(json.dumps(metrics))
except:
disconnected.append(connection)
# 🧹 Clean up
for conn in disconnected:
self.disconnect(conn)
# ⏱️ Update every second
await asyncio.sleep(1)
dashboard_manager = DashboardManager()
@app.websocket("/ws/dashboard")
async def dashboard_websocket(websocket: WebSocket):
await dashboard_manager.connect(websocket)
try:
# 🔄 Keep connection alive
while True:
await websocket.receive_text()
except:
dashboard_manager.disconnect(websocket)🚀 Advanced Concepts
🧙♂️ Advanced Topic 1: Authentication & Security
When you’re ready to level up, add WebSocket authentication:
# 🔐 Secure WebSocket with authentication
from fastapi import FastAPI, WebSocket, Query, WebSocketDisconnect
from fastapi.exceptions import WebSocketException
import jwt
app = FastAPI()
SECRET_KEY = "your-secret-key-here-🔑"
async def verify_token(token: str) -> str:
try:
# 🔍 Verify JWT token
payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
return payload.get("user_id")
except:
raise WebSocketException(code=1008, reason="Invalid authentication")
@app.websocket("/ws/secure")
async def secure_websocket(
websocket: WebSocket,
token: str = Query(...) # 🎫 Token required
):
# 🛡️ Verify authentication
user_id = await verify_token(token)
await websocket.accept()
await websocket.send_text(f"✅ Welcome {user_id}! You're authenticated 🔐")
try:
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Secure echo: {data} 🛡️")
except WebSocketDisconnect:
print(f"🔒 Secure connection closed for {user_id}")🏗️ Advanced Topic 2: Broadcasting with Rooms
For the brave developers, here’s room-based broadcasting:
# 🏠 Room-based WebSocket system
from typing import Dict, Set
from fastapi import FastAPI, WebSocket
import json
app = FastAPI()
class RoomManager:
def __init__(self):
# 🚪 Rooms with their connections
self.rooms: Dict[str, Set[WebSocket]] = {}
async def join_room(self, room_id: str, websocket: WebSocket):
# 🏠 Create room if doesn't exist
if room_id not in self.rooms:
self.rooms[room_id] = set()
self.rooms[room_id].add(websocket)
await self.room_broadcast(room_id, {
"type": "user_joined",
"message": "A new user joined! 🎉",
"room_members": len(self.rooms[room_id])
})
async def leave_room(self, room_id: str, websocket: WebSocket):
if room_id in self.rooms:
self.rooms[room_id].discard(websocket)
# 🧹 Clean empty rooms
if not self.rooms[room_id]:
del self.rooms[room_id]
else:
await self.room_broadcast(room_id, {
"type": "user_left",
"message": "A user left 👋",
"room_members": len(self.rooms[room_id])
})
async def room_broadcast(self, room_id: str, message: dict):
if room_id in self.rooms:
# 📢 Send to everyone in the room
disconnected = set()
for connection in self.rooms[room_id]:
try:
await connection.send_text(json.dumps(message))
except:
disconnected.add(connection)
# 🧹 Remove disconnected
self.rooms[room_id] -= disconnected⚠️ Common Pitfalls and Solutions
😱 Pitfall 1: Forgetting to Handle Disconnections
# ❌ Wrong way - no disconnect handling!
@app.websocket("/ws")
async def bad_websocket(websocket: WebSocket):
await websocket.accept()
while True: # 💥 This will crash on disconnect!
data = await websocket.receive_text()
await websocket.send_text(data)
# ✅ Correct way - handle disconnections gracefully!
@app.websocket("/ws")
async def good_websocket(websocket: WebSocket):
await websocket.accept()
try:
while True:
data = await websocket.receive_text()
await websocket.send_text(data)
except WebSocketDisconnect:
print("👋 Client disconnected gracefully")
except Exception as e:
print(f"⚠️ Error: {e}")🤯 Pitfall 2: Blocking Operations in WebSocket Handlers
# ❌ Dangerous - blocks all connections!
import time
@app.websocket("/ws")
async def blocking_websocket(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_text()
time.sleep(5) # 💥 This blocks everything!
await websocket.send_text("Response")
# ✅ Safe - use async operations!
import asyncio
@app.websocket("/ws")
async def async_websocket(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_text()
await asyncio.sleep(5) # ✅ Non-blocking!
await websocket.send_text("Response")🛠️ Best Practices
- 🎯 Always Handle Disconnections: Use try-except for WebSocketDisconnect
- 📝 Use Connection Managers: Organize connections for scalability
- 🛡️ Implement Authentication: Secure your WebSocket endpoints
- 🎨 Send Structured Data: Use JSON for complex messages
- ✨ Keep Connections Alive: Implement heartbeat/ping-pong mechanism
🧪 Hands-On Exercise
🎯 Challenge: Build a Collaborative Drawing App
Create a real-time collaborative drawing application:
📋 Requirements:
- ✅ Multiple users can draw simultaneously
- 🎨 Different colors for each user
- 👥 Show active users list
- 📍 Broadcast cursor positions
- 🔄 Clear canvas option
🚀 Bonus Points:
- Add drawing tools (pen, eraser, shapes)
- Implement undo/redo functionality
- Save and load drawings
💡 Solution
🔍 Click to see solution
# 🎨 Collaborative drawing app backend
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import Dict, List
import json
import uuid
app = FastAPI()
class DrawingRoom:
def __init__(self):
# 🎨 Active artists
self.artists: Dict[str, WebSocket] = {}
# 📝 Drawing history
self.drawing_history: List[dict] = []
# 🎨 User colors
self.user_colors: Dict[str, str] = {}
async def add_artist(self, websocket: WebSocket) -> str:
artist_id = str(uuid.uuid4())
self.artists[artist_id] = websocket
# 🎨 Assign random color
import random
colors = ["#FF6B6B", "#4ECDC4", "#45B7D1", "#96CEB4", "#FFEAA7"]
self.user_colors[artist_id] = random.choice(colors)
# 📤 Send initial data
await websocket.send_text(json.dumps({
"type": "init",
"artist_id": artist_id,
"color": self.user_colors[artist_id],
"history": self.drawing_history,
"active_users": len(self.artists)
}))
# 📢 Announce new artist
await self.broadcast({
"type": "user_joined",
"message": f"New artist joined! 🎨",
"active_users": len(self.artists)
}, exclude=artist_id)
return artist_id
async def remove_artist(self, artist_id: str):
if artist_id in self.artists:
del self.artists[artist_id]
del self.user_colors[artist_id]
await self.broadcast({
"type": "user_left",
"message": f"Artist left 👋",
"active_users": len(self.artists)
})
async def handle_drawing(self, artist_id: str, data: dict):
# 🎨 Add artist info to drawing data
data["artist_id"] = artist_id
data["color"] = self.user_colors[artist_id]
# 📝 Store in history
if data["type"] == "draw":
self.drawing_history.append(data)
# 📢 Broadcast to all other artists
await self.broadcast(data, exclude=artist_id)
async def clear_canvas(self):
self.drawing_history = []
await self.broadcast({
"type": "clear",
"message": "Canvas cleared! 🧹"
})
async def broadcast(self, data: dict, exclude: str = None):
disconnected = []
for artist_id, websocket in self.artists.items():
if artist_id != exclude:
try:
await websocket.send_text(json.dumps(data))
except:
disconnected.append(artist_id)
# 🧹 Clean up
for artist_id in disconnected:
await self.remove_artist(artist_id)
# 🎯 Create drawing room
drawing_room = DrawingRoom()
@app.websocket("/ws/draw")
async def drawing_websocket(websocket: WebSocket):
await websocket.accept()
artist_id = await drawing_room.add_artist(websocket)
try:
while True:
# 📨 Receive drawing data
data = await websocket.receive_text()
drawing_data = json.loads(data)
if drawing_data["type"] == "clear":
await drawing_room.clear_canvas()
else:
await drawing_room.handle_drawing(artist_id, drawing_data)
except WebSocketDisconnect:
await drawing_room.remove_artist(artist_id)🎓 Key Takeaways
You’ve learned so much! Here’s what you can now do:
- ✅ Create WebSocket endpoints with FastAPI 💪
- ✅ Handle real-time communication between server and clients 🛡️
- ✅ Build connection managers for scalable applications 🎯
- ✅ Implement authentication for secure WebSockets 🐛
- ✅ Create amazing real-time features like chat, dashboards, and collaborative tools! 🚀
Remember: WebSockets open up a world of real-time possibilities! They’re perfect for building interactive, engaging applications. 🤝
🤝 Next Steps
Congratulations! 🎉 You’ve mastered FastAPI WebSockets!
Here’s what to do next:
- 💻 Build the collaborative drawing app from the exercise
- 🏗️ Add WebSockets to an existing FastAPI project
- 📚 Explore WebSocket subprotocols and advanced features
- 🌟 Create your own real-time application!
Remember: Real-time features can transform user experience. Keep experimenting, keep building, and most importantly, have fun creating live, interactive applications! 🚀
Happy real-time coding! 🎉🚀✨