Why WebSockets enable real-time communication in FastAPI - See It in Action

Why WebSockets Enable Real-Time Communication

📖 Scenario: You are building a simple chat server using FastAPI. You want to understand how WebSockets help send messages instantly between clients without waiting for page reloads.

🎯 Goal: Create a FastAPI app that sets up a WebSocket endpoint to accept connections and broadcast messages to all connected clients in real time.

📋 What You'll Learn

Create a FastAPI app instance named app

Create a list called connections to store active WebSocket connections

Add a WebSocket route at /ws that accepts connections

Implement logic to receive messages from one client and send them to all connected clients immediately

💡 Why This Matters

🌍 Real World

WebSockets are used in chat apps, live notifications, online games, and any app needing instant updates without page reloads.

💼 Career

Understanding WebSockets is key for backend developers building real-time features and for full-stack developers integrating live user experiences.

1

Create FastAPI app and connections list

Create a FastAPI app instance called app and a list called connections to hold active WebSocket connections.

FastAPI

from fastapi import FastAPI, WebSocket

# Your code here

Need a hint?

Use app = FastAPI() to create the app and connections = [] to store WebSocket clients.

2

Add WebSocket endpoint at /ws

Add a WebSocket route at /ws using @app.websocket("/ws") and define an async function websocket_endpoint that accepts a websocket parameter of type WebSocket. Inside, accept the connection with await websocket.accept().

FastAPI

from fastapi import FastAPI, WebSocket

app = FastAPI()
connections = []

# Your code here

Need a hint?

Use @app.websocket("/ws") decorator and inside the function call await websocket.accept() to accept the connection.

3

Add connection to list and broadcast messages

Inside websocket_endpoint, add the websocket to the connections list. Then create a while True loop to receive messages with data = await websocket.receive_text(). For each message, send it to all websockets in connections using await connection.send_text(data).

FastAPI

from fastapi import FastAPI, WebSocket

app = FastAPI()
connections = []

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    # Your code here

Need a hint?

Remember to add the websocket to connections and use a loop to receive and broadcast messages.

4

Handle disconnects and remove connections

Wrap the message receiving and sending loop in a try block. Use except WebSocketDisconnect to catch disconnects and remove the websocket from connections with connections.remove(websocket).

FastAPI

from fastapi import FastAPI, WebSocket
from fastapi.websockets import WebSocketDisconnect

app = FastAPI()
connections = []

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    connections.append(websocket)
    # Your code here

Need a hint?

Use try and except WebSocketDisconnect to handle client disconnects and remove them from the list.