WebSockets provide full-duplex communication between a client and a server over a single TCP connection. Unlike traditional HTTP requests, WebSockets allow real-time, bidirectional communication, making them ideal for:
✔ Chat applications
✔ Live notifications
✔ Real-time data streaming
✔ Multiplayer gaming
Installing WebSocket Libraries
To work with WebSockets in Python, install websockets:
pip install websockets
1. Creating a WebSocket Server
A WebSocket server listens for connections and handles messages.
Simple WebSocket Server
import asyncio
import websockets
async def echo(websocket, path):
async for message in websocket:
await websocket.send(f"Server received: {message}")
# Start the WebSocket server on ws://localhost:8765
start_server = websockets.serve(echo, "localhost", 8765)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()
✔ websockets.serve() → Starts the WebSocket server.
✔ async for message in websocket → Listens for incoming messages.
✔ await websocket.send() → Sends messages back to the client.
2. Creating a WebSocket Client
A WebSocket client connects to the server and exchanges messages.
import asyncio
import websockets
async def connect():
async with websockets.connect("ws://localhost:8765") as websocket:
await websocket.send("Hello Server!") # Send message
response = await websocket.recv() # Receive response
print(f"Server Response: {response}")
asyncio.run(connect())
✔ Connects to ws://localhost:8765.
✔ Sends "Hello Server!" and receives a response.
3. Handling Multiple Clients
Modify the server to handle multiple clients.
import asyncio
import websockets
connected_clients = set()
async def handle_client(websocket, path):
connected_clients.add(websocket)
try:
async for message in websocket:
for client in connected_clients:
if client != websocket:
await client.send(message) # Broadcast message
finally:
connected_clients.remove(websocket)
start_server = websockets.serve(handle_client, "localhost", 8765)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()
✔ connected_clients stores all active clients.
✔ Broadcasts received messages to all except the sender.
4. WebSockets with FastAPI
If using FastAPI, WebSockets can be handled with ease.
pythonCopyEditfrom fastapi import FastAPI, WebSocket
from typing import List
app = FastAPI()
clients: List[WebSocket] = []
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
clients.append(websocket)
try:
while True:
data = await websocket.receive_text()
for client in clients:
await client.send_text(f"Client says: {data}")
except:
clients.remove(websocket)
✔ FastAPI provides a websocket route at /ws.
✔ Clients are stored in a list and broadcast messages.
Connecting a WebSocket Client
Use JavaScript in the browser:
let socket = new WebSocket("ws://localhost:8000/ws");
socket.onmessage = function(event) {
console.log("Message from server:", event.data);
};
socket.onopen = function() {
socket.send("Hello from WebSocket client!");
};
✔ Connects to FastAPI WebSocket.
✔ Sends and receives messages.
5. WebSockets with Django (Django Channels)
Django doesn’t support WebSockets natively, so we use Django Channels.
Install Django Channels
pip install channels
Update settings.py
INSTALLED_APPS = [
"django.contrib.admin",
"django.contrib.auth",
"channels",
]
ASGI_APPLICATION = "myproject.asgi.application"
Create a WebSocket Consumer
In consumers.py:
import json
from channels.generic.websocket import AsyncWebsocketConsumer
class ChatConsumer(AsyncWebsocketConsumer):
async def connect(self):
await self.accept()
async def receive(self, text_data):
data = json.loads(text_data)
await self.send(json.dumps({"message": data["message"]}))
✔ Handles WebSocket connect, receive, and send.
Route WebSockets in routing.py
from django.urls import re_path
from .consumers import ChatConsumer
websocket_urlpatterns = [
re_path(r'ws/chat/', ChatConsumer.as_asgi()),
]
Start Django ASGI Server
daphne -b 0.0.0.0 -p 8000 myproject.asgi:application
✔ WebSockets now work in Django! 🎉
6. WebSockets for Real-Time Notifications
Use WebSockets to send instant notifications to users.
WebSocket Server for Notifications
import asyncio
import websockets
async def notify_clients():
async with websockets.connect("ws://localhost:8765") as websocket:
while True:
await asyncio.sleep(5)
await websocket.send("🔔 New Notification!")
asyncio.run(notify_clients())
✔ Sends a notification every 5 seconds.
✔ Perfect for live updates & alerts.
7. WebSockets with Authentication
Protect WebSockets with authentication (JWT, tokens).
Token-Based Authentication in FastAPI
from fastapi import WebSocket, Depends
async def get_token(token: str):
if token != "mysecrettoken":
raise Exception("Unauthorized")
@app.websocket("/ws/{token}")
async def websocket_endpoint(websocket: WebSocket, token: str = Depends(get_token)):
await websocket.accept()
await websocket.send_text("✅ Authentication successful!")
✔ Checks token before accepting connection.
✔ Only users with valid tokens can connect.
8. Handling WebSocket Disconnections
Handle cases where clients disconnect unexpectedly.
async def handle_client(websocket, path):
try:
async for message in websocket:
print(f"Received: {message}")
except websockets.exceptions.ConnectionClosed:
print("Client disconnected")
✔ Detects connection loss and handles errors.
9. Scaling WebSockets with Redis
Use Redis for scalable WebSockets across multiple servers.
Install Redis & Channels Redis
pip install channels-redis
Configure settings.py for Redis
CHANNEL_LAYERS = {
"default": {
"BACKEND": "channels_redis.core.RedisChannelLayer",
"CONFIG": {
"hosts": [("127.0.0.1", 6379)],
},
},
}
✔ Enables WebSocket messaging across multiple servers.
10. WebSockets vs REST APIs
| Feature | WebSockets | REST APIs |
|---|---|---|
| Communication | Bidirectional | Request-Response |
| Latency | Low (Real-time) | Higher |
| Best For | Chats, Live Data | Standard API Calls |
| Performance | More Efficient | Less Efficient |
| Scalability | Needs Redis | Easier to Scale |