FastAPI WebSockets：构建实时聊天应用程序
FastAPI 是一个现代、快速（高性能）的 Web 框架，用于基于标准 Python 类型提示构建 API。其强大功能之一是支持 WebSockets，这使得客户端和服务器之间的实时通信成为可能。在本文中，我们将通过构建一个简单的实时聊天应用程序，探讨如何在 FastAPI 中使用 WebSockets。
1. WebSockets 简介
WebSockets 提供了一个通过单个长连接实现全双工通信的通道。与传统的 HTTP 请求不同，WebSockets 支持双向通信，允许服务器将更新直接发送到客户端，而无需客户端请求。这使得 WebSockets 成为聊天应用、实时通知和游戏等实时应用的理想选择。
2. 设置 FastAPI 项目
首先，安装 FastAPI 和 ASGI 服务器（例如 uvicorn）：
pip install fastapi uvicorn
接着，创建一个新的 Python 文件 main.py，并设置基本的 FastAPI 应用：
from fastapi import FastAPI, WebSocket
from fastapi.websockets import WebSocketDisconnect
app = FastAPI()
@app.get("/")
async def get():
return {"message": "Welcome to FastAPI WebSockets Chat App"}
3. 实现 WebSocket 端点
接下来，我们为聊天应用实现 WebSocket 端点，并创建一个类来管理客户端连接以及向所有连接的客户端广播消息。
from typing import List
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 send_personal_message(self, message: str, websocket: WebSocket):
await websocket.send_text(message)
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
manager = ConnectionManager()
@app.websocket("/ws/chat")
async def websocket_endpoint(websocket: WebSocket):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
await manager.broadcast(f"Client says: {data}")
except WebSocketDisconnect:
manager.disconnect(websocket)
await manager.broadcast("A client disconnected")
4. 创建前端
现在，让我们创建一个简单的 HTML 文件作为聊天应用的前端。在 main.py 的同一目录中创建一个名为 index.html 的新文件。
<!DOCTYPE html>
<html>
<head>
<title>FastAPI Chat</title>
</head>
<body>
<h1>FastAPI WebSockets Chat</h1>
<div id="chat-log"></div>
<input type="text" id="message-input" autocomplete="off"/>
<button onclick="sendMessage()">Send</button>
<script>
const ws = new WebSocket("ws://localhost:8000/ws/chat");
ws.onmessage = function(event) {
const messages = document.getElementById('chat-log');
const message = document.createElement('div');
const content = document.createTextNode(event.data);
message.appendChild(content);
messages.appendChild(message);
};
function sendMessage() {
const input = document.getElementById("message-input");
ws.send(input.value);
input.value = '';
}
</script>
</body>
</html>
5. 运行和测试应用程序
要运行应用程序，请使用 uvicorn 启动 FastAPI 服务器：
uvicorn main:app --reload
打开 Web 浏览器并导航到 http://localhost:8000，查看聊天应用程序的运行情况。您可以打开多个浏览器窗口或选项卡，在它们之间实时发送消息。
6. 增强功能
1. 添加用户身份验证
可以通过在前端添加用户名输入来增强聊天应用。更新前端 HTML：
<input type="text" id="username-input" placeholder="Enter your username" autocomplete="off"/>
并修改 sendMessage() 函数：
function sendMessage() {
const input = document.getElementById("message-input");
const username = document.getElementById("username-input").value;
ws.send(`${username}: ${input.value}`);
input.value = '';
}
后端不需要做太多更改，依旧会接收并广播完整消息。
2. 消息持久化
为了持久化消息，可以使用 SQLite 数据库来存储聊天记录，并在客户端连接时加载历史记录。
在 main.py 中添加 SQLite 数据库的相关代码：
import sqlite3
# Database setup
conn = sqlite3.connect('chat.db')
c = conn.cursor()
c.execute('''CREATE TABLE IF NOT EXISTS messages (content TEXT)''')
conn.commit()
# 在广播消息时，保存到数据库
async def broadcast(self, message: str):
c.execute("INSERT INTO messages (content) VALUES (?)", (message,))
conn.commit()
for connection in self.active_connections:
await connection.send_text(message)
3. 添加输入指示器
为了显示用户正在输入的消息，可以扩展前端，使用 WebSocket.send() 发送 "用户正在输入" 消息。
<input type="text" id="message-input" autocomplete="off" oninput="indicateTyping()"/>
<script>
function indicateTyping() {
const username = document.getElementById("username-input").value;
ws.send(`${username} is typing...`);
}
</script>
后端无需做太多修改，直接接收并广播这些“输入中”的提示消息。
结语
本文展示了如何使用 FastAPI 和 WebSockets 构建一个实时聊天应用。我们实现了基本的聊天功能，并逐步扩展了身份验证、消息持久化和输入指示器等功能。WebSockets 是构建实时 Web 应用程序的理想工具，FastAPI 简洁的架构使其实现非常简单。