The Ultimate Guide to FastAPI WebSocket Examples

FastAPI is a modern, fast, web framework for building APIs with Python 3.7+ based on standard Python type hints. One of the many features that make FastAPI stand out is its ability to handle WebSocket connections, which allows for real-time communication between clients and servers. In this article, we will explore the best FastAPI WebSocket examples and how you can use them to enhance your web applications.

What is FastAPI WebSocket?

WebSocket is a protocol that enables two-way communication between a client and a server over a single, long-lived connection. Unlike traditional HTTP requests, which are stateless and require a new connection each time, WebSocket connections remain open, allowing real-time communication and reducing latency.

FastAPI WebSocket is an implementation of the WebSocket protocol in the FastAPI framework. It allows you to create real-time applications that can handle multiple connections simultaneously and communicate with clients in real-time.

Why use FastAPI WebSocket?

FastAPI WebSocket is an excellent choice for building real-time applications that require fast communication between clients and servers. It offers several advantages over traditional HTTP requests, including:

  • Reduced latency: WebSocket connections remain open, allowing real-time communication and reducing latency.
  • Increased efficiency: WebSocket connections use less bandwidth than traditional HTTP requests, reducing server load and improving performance.
  • Real-time updates: With WebSocket, you can push real-time updates to clients without requiring them to refresh the page.

FastAPI WebSocket Example: Chat Application

One of the most popular FastAPI WebSocket examples is a chat application. This example demonstrates how to use FastAPI WebSocket to create a real-time chat application that allows users to communicate with each other in real-time.

The chat application consists of two parts: the server-side code and the client-side code. The server-side code handles incoming WebSocket connections and broadcasts messages to all connected clients. The client-side code creates a WebSocket connection and sends and receives messages from the server.

Server-Side Code

To create the server-side code for the chat application, you will need to install the following packages:

  1. fastapi: The FastAPI framework.
  2. uvicorn: A lightning-fast ASGI server.
  3. websockets: A Python implementation of the WebSocket protocol.

Once you have installed these packages, you can create the server-side code for the chat application:

from fastapi import FastAPI, WebSocketfrom fastapi.responses import HTMLResponsefrom fastapi.staticfiles import StaticFilesfrom fastapi.templating import Jinja2Templatesfrom typing import List

app = FastAPI()

app.mount("/static", StaticFiles(directory="static"), name="static")

templates = Jinja2Templates(directory="templates")

class ConnectionManager:def __init__(self):self.active_connections: List[WebSocket] = []

async def connect(self, websocket: WebSocket):await websocket.accept()self.active_connections.append(websocket)

def disconnect(self, websocket: WebSocket):self.active_connections.remove(websocket)

async def broadcast(self, message: str):for connection in self.active_connections:await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/{client_id}")async def websocket_endpoint(websocket: WebSocket, client_id: int):await manager.connect(websocket)try:while True:data = await websocket.receive_text()await manager.broadcast(f"Client {client_id}: {data}")except WebSocketDisconnect:manager.disconnect(websocket)await manager.broadcast(f"Client {client_id} left the chat")

This code creates a FastAPI application and defines a WebSocket endpoint at “/ws/{client_id}”. The WebSocket endpoint accepts a “client_id” parameter, which is used to identify the client. When a client connects to the WebSocket endpoint, the “connect” method of the ConnectionManager class is called, which adds the WebSocket connection to the list of active connections.

The WebSocket endpoint then enters a loop, waiting for incoming messages from the client. When a message is received, the “broadcast” method of the ConnectionManager class is called, which sends the message to all connected clients. If a client disconnects from the WebSocket endpoint, the “disconnect” method of the ConnectionManager class is called, which removes the WebSocket connection from the list of active connections.

Client-Side Code

To create the client-side code for the chat application, you will need to create an HTML file that contains a form for sending messages and a div for displaying messages. You will also need to include the following JavaScript code:

const chat = document.getElementById("chat");const form = document.getElementById("form");const input = document.getElementById("input");

const websocket = new WebSocket(`ws://${window.location.host}/ws/${Math.floor(Math.random() * 1000)}`);

websocket.onmessage = (event) => {const message = event.data;const p = document.createElement("p");p.innerText = message;chat.appendChild(p);};

form.addEventListener("submit", (event) => {event.preventDefault();const message = input.value;input.value = "";websocket.send(message);});

This code creates a WebSocket connection to the server-side WebSocket endpoint and defines an event listener for incoming messages. When a message is received, the code creates a new <p> element and appends it to the chat div. The code also defines an event listener for the form submit event, which sends the message to the server-side WebSocket endpoint.

FastAPI WebSocket Example: Real-Time Dashboard

Another popular FastAPI WebSocket example is a real-time dashboard. This example demonstrates how to use FastAPI WebSocket to create a real-time dashboard that displays real-time data from a database or API.

The real-time dashboard consists of two parts: the server-side code and the client-side code. The server-side code retrieves data from a database or API and sends it to the client-side code via WebSocket. The client-side code displays the data in real-time.

Server-Side Code

To create the server-side code for the real-time dashboard, you will need to install the following packages:

  1. fastapi: The FastAPI framework.
  2. uvicorn: A lightning-fast ASGI server.
  3. websockets: A Python implementation of the WebSocket protocol.
  4. sqlalchemy: A Python SQL toolkit and ORM.
  5. alembic: A database migration tool for SQLAlchemy.

Once you have installed these packages, you can create the server-side code for the real-time dashboard:

from fastapi import FastAPI, WebSocketfrom fastapi.responses import HTMLResponsefrom fastapi.staticfiles import StaticFilesfrom fastapi.templating import Jinja2Templatesfrom sqlalchemy import create_engine, Column, Integer, String, Float, DateTimefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import sessionmakerfrom typing import Listfrom datetime import datetime, timedeltaimport random

app = FastAPI()

app.mount("/static", StaticFiles(directory="static"), name="static")

templates = Jinja2Templates(directory="templates")

engine = create_engine("sqlite:///data.db")SessionLocal = sessionmaker(bind=engine)Base = declarative_base()

class DataPoint(Base):__tablename__ = "datapoints"id = Column(Integer, primary_key=True, index=True)timestamp = Column(DateTime, default=datetime.utcnow())value = Column(Float, default=random.random())

Base.metadata.create_all(bind=engine)

def get_db():db = SessionLocal()try:yield dbfinally:db.close()

class ConnectionManager:def __init__(self):self.active_connections: List[WebSocket] = []

async def connect(self, websocket: WebSocket):await websocket.accept()self.active_connections.append(websocket)

def disconnect(self, websocket: WebSocket):self.active_connections.remove(websocket)

async def broadcast(self, message: str):for connection in self.active_connections:await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws")async def websocket_endpoint(websocket: WebSocket):await manager.connect(websocket)try:while True:db = next(get_db())data = db.query(DataPoint).order_by(DataPoint.timestamp.desc()).limit(10).all()data.reverse()chart_data = [{"timestamp": dp.timestamp.strftime("%Y-%m-%d %H:%M:%S"), "value": dp.value} for dp in data]await websocket.send_json(chart_data)await asyncio.sleep(1)except WebSocketDisconnect:manager.disconnect(websocket)

This code creates a FastAPI application and defines a WebSocket endpoint at “/ws”. The WebSocket endpoint retrieves the last 10 data points from a SQLite database and sends them to the client-side code via WebSocket. The WebSocket endpoint then waits for one second before repeating the process.

Client-Side Code

To create the client-side code for the real-time dashboard, you will need to create an HTML file that contains a div for displaying the data. You will also need to include the following JavaScript code:

const chart = document.getElementById("chart");

const websocket = new WebSocket(`ws://${window.location.host}/ws`);

websocket.onmessage = (event) => {const chartData = JSON.parse(event.data);const labels = chartData.map((dp) => dp.timestamp);const data = chartData.map((dp) => dp.value);

const ctx = chart.getContext("2d");const chartInstance = new Chart(ctx, {type: "line",data: {labels: labels,datasets: [{label: "Real-Time Data",data: data,backgroundColor: "rgba(255, 99, 132, 0.2)",borderColor: "rgba(255, 99, 132, 1)",borderWidth: 1,},],},options: {scales: {xAxes: [{type: "time",time: {unit: "second",displayFormats: {second: "h:mm:ss a",},},distribution: "linear",},],},},});};

This code creates a WebSocket connection to the server-side WebSocket endpoint and defines an event listener for incoming messages. When a message is received, the code creates a new Chart.js instance and uses the data to create a real-time line chart.

FAQ

What is FastAPI?

FastAPI is a modern, fast, web framework for building APIs with Python 3.7+ based on standard Python type hints. It is designed to be easy to use, fast, and scalable.

What is WebSocket?

WebSocket is a protocol that enables two-way communication between a client and a server over a single, long-lived connection. Unlike traditional HTTP requests, which are stateless and require a new connection each time, WebSocket connections remain open, allowing real-time communication and reducing latency.

What are the advantages of FastAPI WebSocket?

FastAPI WebSocket offers several advantages over traditional HTTP requests, including reduced latency, increased efficiency, and real-time updates.

What are some popular FastAPI WebSocket examples?

Some popular FastAPI WebSocket examples include a chat application, a real-time dashboard, and a real-time game.

What packages do I need to use FastAPI WebSocket?

To use FastAPI WebSocket, you will need to install the fastapi, uvicorn, and websockets packages.

Can I use FastAPI WebSocket with a database or API?

Yes, you can use FastAPI WebSocket with a database or API. The real-time dashboard example demonstrates how to retrieve data from a SQLite database and send it to the client-side code via WebSocket.

Is FastAPI WebSocket suitable for large-scale applications?

Yes, FastAPI WebSocket is suitable for large-scale applications. It is designed to be fast, efficient, and scalable, making it an excellent choice for building real-time applications that require high performance.