What is FastAPI WebSocket?
FastAPI is a modern, fast (high-performance) web framework for building APIs with Python 3.6+ based on standard Python type hints. One of the powerful features of FastAPI is its support for WebSockets, which are crucial for real-time communication applications. Unlike traditional HTTP connections,
WebSockets
allow for two-way interactive communication sessions between the user's browser and a server. This guide aims to provide a step-by-step tutorial on implementing WebSockets in FastAPI, covering everything from the initial setup to advanced use cases.Setting Up FastAPI
To get started with FastAPI, you need to install it along with Uvicorn, which is an ASGI server implementation for fast asynchronous web applications. You can install both using pip:
bash
1pip install fastapi uvicorn
2Once installed, you can set up a basic FastAPI project. Create a new directory for your project and inside it, create a main.py file:
Python
1# main.py
2from fastapi import FastAPI
3
4app = FastAPI()
5
6@app.get("/")
7async def read_root():
8 return {"Hello": "World"}
9To run the application, use the following command:
bash
1uvicorn main:app --reload
2This will start the FastAPI server with auto-reload enabled. You can now access your application at
http://127.0.0.1:8000
.Understanding WebSockets
WebSockets are a protocol that provides full-duplex communication channels over a single TCP connection. Unlike HTTP, which is a request-response protocol, WebSockets allow for persistent connections where both the client and the server can send messages to each other at any time. This makes WebSockets ideal for applications that require real-time updates, such as chat applications, live notifications, and
online gaming
.Implementing WebSockets in FastAPI
To implement WebSockets in FastAPI, you need to define a WebSocket endpoint. Here's a basic example of an echo server, which sends back any message it receives from the client:
Python
1from fastapi import FastAPI, WebSocket
2
3app = FastAPI()
4
5@app.websocket("/ws")
6async def websocket_endpoint(websocket: WebSocket):
7 await websocket.accept()
8 while True:
9 data = await websocket.receive_text()
10 await websocket.send_text(f"Message text was: {data}")
11In this example, the server accepts the WebSocket connection, waits for messages from the client, and sends back the received message.
Broadcasting Messages
Handling multiple clients in a WebSocket server involves maintaining a list of active connections and broadcasting messages to all connected clients. Here’s an example:
Python
1from typing import List
2
3clients: List[WebSocket] = []
4
5@app.websocket("/ws")
6async def websocket_endpoint(websocket: WebSocket):
7 await websocket.accept()
8 clients.append(websocket)
9 try:
10 while True:
11 data = await websocket.receive_text()
12 for client in clients:
13 await client.send_text(f"Message text was: {data}")
14 except WebSocketDisconnect:
15 clients.remove(websocket)
16In this code, we keep track of all connected clients in the
clients list and broadcast any received message to all of them.Handling WebSocket Disconnects
Properly managing client disconnections is crucial for maintaining a stable WebSocket server. Use the
WebSocketDisconnect exception to handle client disconnects:Python
1from fastapi import WebSocketDisconnect
2
3@app.websocket("/ws")
4async def websocket_endpoint(websocket: WebSocket):
5 await websocket.accept()
6 clients.append(websocket)
7 try:
8 while True:
9 data = await websocket.receive_text()
10 for client in clients:
11 await client.send_text(f"Message text was: {data}")
12 except WebSocketDisconnect:
13 clients.remove(websocket)
14This code ensures that disconnected clients are removed from the
clients list, preventing errors.Advanced WebSocket Features
For more advanced WebSocket features, such as implementing authentication and securing WebSocket connections, you can use dependencies like OAuth2. Here’s an example:
Python
1from fastapi import Depends, WebSocket, status
2from fastapi.exceptions import HTTPException
3from fastapi.security import OAuth2PasswordBearer
4
5oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
6
7async def get_current_user(token: str = Depends(oauth2_scheme)):
8 # Your authentication logic here
9 return token
10
11@app.websocket("/ws")
12async def websocket_endpoint(websocket: WebSocket, token: str = Depends(get_current_user)):
13 try:
14 await websocket.accept()
15 # Your WebSocket logic here
16 except HTTPException as e:
17 await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
18In this example, the
get_current_user function checks the token for authentication. If the token is invalid, the WebSocket connection is closed with a policy violation status.Common Use Cases for FastAPI WebSockets
FastAPI WebSockets are versatile and can be used in various real-time applications:
- Real-time Chat Applications: Enable users to communicate in real time, sending and receiving messages instantly.
- Live Notifications: Provide users with instant updates about events, such as new messages, alerts, or data changes.
- Collaborative Editing: Allow multiple users to work on the same document simultaneously, with changes being reflected in real time.
These use cases demonstrate the flexibility and power of WebSockets in creating interactive and engaging applications.
Conclusion
FastAPI's support for WebSockets provides developers with a powerful tool for creating real-time, interactive applications. By following this guide, you've learned how to set up a FastAPI project, implement basic and advanced WebSocket functionality, handle client connections and disconnections, and secure your WebSocket connections with authentication.
Whether you're building a chat application, live notification system, or any other real-time service, FastAPI and WebSockets offer the performance and simplicity needed to achieve your goals. Continue exploring and experimenting with FastAPI WebSockets to unlock their full potential in your projects.
FAQ