Today we are going to learn how we can quickly and easily generate a websocket with FastAPI.
Step 1: Environment Setup, if you already have FastAPI installed, you can skip these steps.
Make sure you have Python installed on your system. Then, install FastAPI and Uvicorn using pip:
pip install fastapi uvicorn
Step 2: Create a FastAPI Project
Create a new directory for your project and inside it, create a file named main.py.
mkdir fastapi_websocket cd fastapi_websocket touch main.py
Step 3: Configure FastAPI Server with WebSocket
Open the main.py file in your favorite text editor and implement the following code:
from fastapi import FastAPI, WebSocket
app = FastAPI()
class ConnectionManager:
def __init__(self):
self.connections = {}
async def connect(self, websocket: WebSocket, client_id: int):
await websocket.accept()
self.connections[client_id] = websocket
def disconnect(self, client_id: int):
del self.connections[client_id]
async def send_message(self, message: str, client_id: int):
await self.connections[client_id].send_text(message)
manager = ConnectionManager()
@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
await manager.connect(websocket, client_id)
try:
while True:
data = await websocket.receive_text()
await manager.send_message(f"Message received: {data}", client_id)
except Exception as e:
print(f"Error: {e}")
finally:
manager.disconnect(client_id)
This code sets up a WebSocket in FastAPI with an endpoint /ws/{client_id}. The ConnectionManager class manages connections and sends messages to clients.
Step 4: Run the FastAPI Server
Open your terminal and run the following command to start the FastAPI server:
uvicorn main:app --reload
This will start the server at http://127.0.0.1:8000.
Step 5: Test the WebSocket with FastAPI on a Domain
You can use this website if your server is online: https://piehost.com/websocket-tester
Enter the URL of your websocket server: wss://yourdomain.com/ws/1
And you’ll receive the following:
