request和websocket

当然，可以为你详细介绍 FastAPI 中的 Request 对象。Request 对象在 FastAPI 中扮演着重要的角色，负责封装来自客户端的 HTTP 请求信息。了解 Request 对象的使用方法和属性，有助于你更高效地处理请求数据、访问请求上下文以及进行各种操作。

1. 什么是 Request 对象

Request 对象是 FastAPI 提供的一个类，用于表示客户端发送到服务器的 HTTP 请求。它基于 Starlette 的 Request 类，提供了丰富的属性和方法来访问请求的各个部分，如路径参数、查询参数、请求体、头信息等。

2. 如何在路由中使用 Request 对象

在 FastAPI 的路由函数中，你可以通过参数注入的方式使用 Request 对象。只需将 Request 作为函数参数，并使用 from fastapi import Request 导入即可。

示例：

from fastapi import FastAPI, Requestapp = FastAPI()@app.get("/items/{item_id}")
async def read_item(request: Request, item_id: int):client_host = request.client.hostreturn {"item_id": item_id, "client_host": client_host}

在这个示例中，Request 对象被注入到 read_item 路由函数中，可以通过它访问请求的详细信息。

3. Request 对象的主要属性

Request 对象包含了大量有用的属性，以下是一些常用的属性：

• url: 请求的完整 URL 对象。

  request.url  # 返回一个 `URL` 对象，包含 scheme、host、path、query 等信息
  

• method: 请求的方法，如 GET、POST、PUT 等。

  request.method  # 例如 'GET'
  

• headers: 请求的头信息，类似于字典。

  user_agent = request.headers.get('user-agent')
  

• query_params: 请求的查询参数，提供类似字典的接口。

  params = request.query_params  # 例如 {'q': 'search', 'page': '2'}
  

• path_params: 路径参数，定义在路由路径中的变量。

  path_param = request.path_params.get('item_id')
  

• cookies: 请求中的 cookies，类似于字典。

  session_id = request.cookies.get('session_id')
  

• client: 客户端的连接信息，包括 IP 地址和端口。

  client_host = request.client.host
  client_port = request.client.port
  

• state: 一个用于在请求生命周期中存储自定义状态信息的属性。

  request.state.some_data = "value"
  

4. Request 对象的主要方法

除了属性，Request 对象还提供了一些方法，用于处理请求体和其他操作：

• json(): 异步方法，用于解析请求体中的 JSON 数据。

  data = await request.json()
  

• form(): 异步方法，用于解析表单数据（application/x-www-form-urlencoded 或 multipart/form-data）。

  form_data = await request.form()
  

• body(): 异步方法，获取原始的请求体内容。

  raw_body = await request.body()
  

• stream(): 异步方法，获取一个异步生成器，用于逐步读取请求体数据。

  async for chunk in request.stream():process(chunk)
  

5. Request 对象与其他组件的关系

Request 对象常与以下组件一起使用：

• 依赖注入（Dependency Injection）: 可以将 Request 对象作为依赖项注入到其他依赖函数中，方便在不同层级访问请求信息。

  from fastapi import Dependsasync def get_user_agent(request: Request):return request.headers.get('user-agent')@app.get("/user-agent")
  async def read_user_agent(user_agent: str = Depends(get_user_agent)):return {"user_agent": user_agent}
  

• 中间件（Middleware）: 中间件可以访问和修改 Request 对象，实现全局请求处理逻辑。

  from fastapi import FastAPI, Request
  from starlette.middleware.base import BaseHTTPMiddlewareapp = FastAPI()class CustomMiddleware(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next):# 在请求处理前执行response = await call_next(request)# 在请求处理后执行response.headers["X-Custom-Header"] = "Custom Value"return responseapp.add_middleware(CustomMiddleware)
  

6. 示例：使用 Request 对象访问请求数据

以下是一个综合示例，展示如何使用 Request 对象访问各种请求数据：

from fastapi import FastAPI, Request, HTTPException
from typing import Optionalapp = FastAPI()@app.post("/submit")
async def submit_form(request: Request):# 获取请求方法method = request.method  # 'POST'# 获取请求 URLurl = str(request.url)  # 'http://localhost:8000/submit'# 获取请求头user_agent = request.headers.get('user-agent')# 获取查询参数query_params = request.query_params  # immutable multi-dict# 获取路径参数（假设路由定义了路径参数）# path_params = request.path_params# 获取 cookiescookies = request.cookies# 解析 JSON 请求体try:data = await request.json()except Exception:data = None# 解析表单数据# form_data = await request.form()return {"method": method,"url": url,"user_agent": user_agent,"query_params": query_params,"cookies": cookies,"data": data,}@app.get("/items/")
async def read_items(request: Request, q: Optional[str] = None):client_host = request.client.hostreturn {"client_host": client_host, "q": q}

在这个示例中：

• /submit 路由接收一个 POST 请求，并通过 Request 对象访问请求方法、URL、头信息、查询参数、cookies 和请求体数据。
• /items/ 路由接收一个 GET 请求，展示如何获取客户端的 IP 地址和查询参数。

7. 注意事项

• 异步操作: Request 对象的方法如 json()、form() 和 body() 是异步方法，必须使用 await 来调用。

  data = await request.json()
  

• 多次读取请求体: 由于请求体是一次性可读取的流，避免在同一个请求中多次读取请求体，否则会导致错误或空数据。

• 性能考虑: 对于大型请求体，避免一次性加载所有数据，可以使用 stream() 方法逐步处理。

• 错误处理: 在解析请求体时，捕获可能的异常，如 JSON 解析错误，防止应用崩溃。

8. Request 对象与 WebSocket 对象的区别

虽然 Request 对象和 WebSocket 对象都用于处理客户端的连接请求，但它们在使用场景和功能上有所不同：

• 使用场景:
  • Request 对象用于传统的 HTTP 请求（如 GET、POST）。
  • WebSocket 对象用于 WebSocket 连接，实现实时的双向通信。
• 功能差异:
  • Request 对象提供访问请求头、路径参数、查询参数、请求体等功能。
  • WebSocket 对象提供发送和接收消息的方法，如 send_text()、receive_text()、send_json()、receive_json() 等。

示例：

from fastapi import FastAPI, Request, WebSocketapp = FastAPI()@app.get("/http-endpoint")
async def http_endpoint(request: Request):data = await request.json()return {"received_data": data}@app.websocket("/ws-endpoint")
async def websocket_endpoint(websocket: WebSocket):await websocket.accept()data = await websocket.receive_text()await websocket.send_text(f"Message text was: {data}")

在这个示例中：

• /http-endpoint 使用 Request 对象处理 HTTP 请求，解析 JSON 请求体。
• /ws-endpoint 使用 WebSocket 对象处理 WebSocket 连接，接收和发送文本消息。

总结

Request 对象是 FastAPI 中处理 HTTP 请求的核心组件，提供了丰富的属性和方法来访问和操作请求数据。通过合理地使用 Request 对象，你可以高效地获取请求信息、处理请求体数据，并与其他组件如依赖注入、中间件等协同工作。此外，了解 Request 对象与 WebSocket 对象的区别，有助于在不同的通信场景中选择合适的处理方式。

当然！下面，我将系统性地讲解如何在 FastAPI 中使用 Request 和 WebSocket 对象来获取请求消息、处理请求体数据等操作，并通过一个实战案例（一个简单的聊天室应用）来展示如何应用这些知识。所有代码将附带详细的注释，帮助你更好地理解每一部分的功能和实现。

---

目录

1. FastAPI 的 Request 对象
   • 访问 HTTP 请求数据
   • 常用方法
   • 示例：在标准路由中使用 Request
2. FastAPI 的 WebSocket 对象
   • 建立 WebSocket 连接
   • 发送和接收消息
   • 管理连接
3. 结合使用 Request 和 WebSocket
   • 通过 JWT 进行认证
   • 在 HTTP 请求和 WebSocket 之间传递上下文
4. 实战案例：构建一个简单的聊天室
   • 项目结构
   • 设置 FastAPI 应用
   • 实现 HTTP 路由使用 Request
   • 实现 WebSocket 端点
   • 处理消息和数据
   • 测试应用
5. 总结

---

1. FastAPI 的 Request 对象

Request 对象是 FastAPI 中用于表示客户端发送的 HTTP 请求的核心组件。它基于 Starlette 的 Request 类，提供了丰富的属性和方法来访问请求的各个部分。

访问 HTTP 请求数据

通过将 Request 对象作为路由函数的参数，可以访问到请求的详细信息，如路径参数、查询参数、请求体、头信息等。

常用方法

• json(): 异步方法，用于解析请求体中的 JSON 数据。

  data = await request.json()
  

• form(): 异步方法，用于解析表单数据（application/x-www-form-urlencoded 或 multipart/form-data）。

  form_data = await request.form()
  

• body(): 异步方法，获取原始的请求体内容。

  raw_body = await request.body()
  

• stream(): 异步方法，获取一个异步生成器，用于逐步读取请求体数据。

  async for chunk in request.stream():process(chunk)
  

示例：在标准路由中使用 Request

以下示例展示了如何在 FastAPI 的 HTTP 路由中使用 Request 对象访问和处理请求数据。

from fastapi import FastAPI, Request, HTTPException
from typing import Optionalapp = FastAPI()@app.post("/submit")
async def submit_form(request: Request):"""处理提交的表单数据。通过 Request 对象访问请求的各种数据，如方法、URL、头信息、查询参数、cookies 和请求体。"""# 获取请求方法，如 'POST'method = request.method# 获取请求的完整 URL，如 'http://localhost:8000/submit'url = str(request.url)# 获取特定的请求头，例如 'User-Agent'user_agent = request.headers.get('user-agent')# 获取所有查询参数，返回一个不可变的多重字典query_params = request.query_params# 获取所有 cookies，返回一个字典cookies = request.cookies# 尝试解析请求体中的 JSON 数据try:data = await request.json()except Exception:# 如果解析失败，设置为 None 或处理异常data = Nonereturn {"method": method,"url": url,"user_agent": user_agent,"query_params": query_params,"cookies": cookies,"data": data,}@app.get("/items/")
async def read_items(request: Request, q: Optional[str] = None):"""读取项目列表，展示如何获取客户端的 IP 地址和查询参数。- `q`: 可选的查询参数，用于过滤项目。"""# 获取客户端的 IP 地址client_host = request.client.hostreturn {"client_host": client_host, "q": q}

---

2. FastAPI 的 WebSocket 对象

WebSocket 对象用于处理实时的双向通信，允许服务器和客户端之间进行持续的数据交换。与传统的 HTTP 请求不同，WebSocket 提供了一个持久的连接，可以在不重新建立连接的情况下发送和接收消息。

建立 WebSocket 连接

通过定义 WebSocket 路由，可以处理客户端的 WebSocket 连接请求。

from fastapi import FastAPI, WebSocket, WebSocketDisconnectapp = FastAPI()@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):"""处理客户端的 WebSocket 连接。建立连接后，持续接收和发送消息，直到客户端断开连接。"""# 接受 WebSocket 连接请求await websocket.accept()try:while True:# 接收来自客户端的文本消息data = await websocket.receive_text()# 将接收到的消息发送回客户端await websocket.send_text(f"Message text was: {data}")except WebSocketDisconnect:# 处理客户端断开连接的情况print("Client disconnected")

发送和接收消息

• 发送消息:

  • send_text(data: str): 发送文本消息。

    await websocket.send_text("Hello, client!")
    

  • send_json(data: Any): 发送 JSON 格式的消息。

    await websocket.send_json({"message": "Hello, client!"})
    

  • send_bytes(data: bytes): 发送字节消息。

    await websocket.send_bytes(b"Hello, client!")
    

• 接收消息:

  • receive_text(): 接收文本消息。

    data = await websocket.receive_text()
    

  • receive_json(): 接收 JSON 格式的消息。

    data = await websocket.receive_json()
    

  • receive_bytes(): 接收字节消息。

    data = await websocket.receive_bytes()
    

管理连接

在 WebSocket 应用中，通常需要管理多个连接，包括：

• 跟踪活跃连接：维护一个连接列表或字典，以便向特定客户端发送消息或广播消息。
• 处理连接生命周期：处理连接的建立、消息传递和断开连接的逻辑。

---

3. 结合使用 Request 和 WebSocket

在实际应用中，可能需要结合使用 Request 和 WebSocket 对象，例如在建立 WebSocket 连接前进行认证，或者在 WebSocket 会话中访问特定的请求上下文信息。

通过 JWT 进行认证

使用 JWT（JSON Web Token）进行认证，可以在建立 WebSocket 连接时验证用户身份。下面是一个示例，展示如何在 WebSocket 连接中使用 JWT 进行认证。

设置认证依赖

首先，配置 JWT 认证相关的依赖。

auth.py:

from fastapi_jwt_auth import AuthJWT
from pydantic import BaseModelclass Settings(BaseModel):authjwt_secret_key: str = "supersecretkey"  # JWT 密钥@AuthJWT.load_config
def get_config():return Settings()

实现 WebSocket 认证

在 WebSocket 路由中，使用 JWT 进行认证。

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Depends, HTTPException
from fastapi_jwt_auth import AuthJWT
from auth import Settingsapp = FastAPI()@app.websocket("/ws/protected")
async def protected_websocket(websocket: WebSocket, Authorize: AuthJWT = Depends()):"""受保护的 WebSocket 端点，只有通过 JWT 认证的用户可以连接。通过 URL 查询参数传递 JWT 令牌。"""try:# 从查询参数中获取 JWT 令牌token = websocket.query_params.get("token")if not token:# 如果没有提供令牌，关闭连接并返回未授权错误await websocket.close(code=1008, reason="Unauthorized: No token provided")return# 使用 JWT 进行认证Authorize.jwt_required(auth_from="websocket", token=token)# 获取 JWT 中的主题（通常是用户名或用户 ID）user = Authorize.get_jwt_subject()# 接受 WebSocket 连接await websocket.accept()# 发送认证成功消息await websocket.send_text(f"Hello {user}, you are authenticated!")# 持续接收和处理消息while True:data = await websocket.receive_text()# 这里可以添加逻辑处理接收到的消息await websocket.send_text(f"{user} says: {data}")except Exception as e:# 处理认证失败或其他异常，关闭连接await websocket.close(code=1008, reason="Unauthorized")

在 HTTP 请求和 WebSocket 之间传递上下文

可以在 HTTP 请求中生成一些上下文信息，并在 WebSocket 连接中使用这些信息。例如，通过 HTTP 请求创建一个会话 ID，然后在 WebSocket 连接中使用该会话 ID 进行消息传递。

from fastapi import FastAPI, Request, WebSocket, Depends
from fastapi.responses import JSONResponse
from typing import Dict
import uuidapp = FastAPI()# 简单的内存存储，用于管理会话和连接
connections: Dict[str, WebSocket] = {}@app.post("/create-session")
async def create_session():"""创建一个新的聊天会话，并返回唯一的会话 ID。"""# 生成唯一的会话 IDsession_id = str(uuid.uuid4())return {"session_id": session_id}@app.websocket("/ws/{session_id}")
async def websocket_endpoint(websocket: WebSocket, session_id: str):"""WebSocket 端点，使用会话 ID 来管理连接。客户端需要通过 URL 提供会话 ID，以建立与特定会话的连接。"""await websocket.accept()# 将 WebSocket 连接与会话 ID 关联connections[session_id] = websockettry:while True:# 接收消息data = await websocket.receive_text()# 处理消息，例如广播给同一会话的其他客户端# 这里简单地将消息回发给发送者await websocket.send_text(f"Session {session_id}: {data}")except WebSocketDisconnect:# 处理客户端断开连接，移除会话关联del connections[session_id]print(f"Session {session_id} disconnected")

---

4. 实战案例：构建一个简单的聊天室

通过一个实战案例，整合使用 Request 和 WebSocket 对象，构建一个简单的聊天室应用。这个聊天室允许用户通过 WebSocket 连接发送和接收消息，同时使用 HTTP 请求进行用户认证。

项目结构

chat_app/
├── main.py
├── dependencies.py
├── schemas.py
├── auth.py
└── requirements.txt

1. 设置 FastAPI 应用

首先，创建项目文件并安装所需依赖。

requirements.txt:

fastapi
uvicorn
fastapi-jwt-auth
pydantic

安装依赖：

pip install -r requirements.txt

2. 定义依赖和认证

创建 auth.py 和 schemas.py 文件，处理用户认证和数据模式。

auth.py:

# auth.pyfrom fastapi_jwt_auth import AuthJWT
from pydantic import BaseModelclass Settings(BaseModel):authjwt_secret_key: str = "supersecretkey"  # 设置 JWT 的密钥authjwt_access_token_expires: int = 60 * 60  # 访问令牌过期时间（秒）@AuthJWT.load_config
def get_config():"""加载 JWT 配置。FastAPI-JWT-Auth 使用这个配置类来获取 JWT 的相关设置。"""return Settings()# 模拟的用户数据库，用于示例认证
fake_users_db = {"user1": {"username": "user1", "password": "password1"},"user2": {"username": "user2", "password": "password2"},
}def authenticate_user(username: str, password: str):"""验证用户凭证。在实际应用中，应从数据库中查询用户并验证密码。"""user = fake_users_db.get(username)if user and user["password"] == password:return userreturn None

schemas.py:

# schemas.pyfrom pydantic import BaseModelclass UserLogin(BaseModel):"""用户登录的请求数据模式。包含用户名和密码字段。"""username: strpassword: strclass Message(BaseModel):"""聊天消息的数据模式。包含发送者的用户名和消息内容。"""username: strmessage: str

3. 创建主应用和路由

main.py:

# main.pyfrom fastapi import FastAPI, WebSocket, WebSocketDisconnect, Depends, HTTPException
from fastapi.responses import HTMLResponse
from fastapi_jwt_auth import AuthJWT
from auth import authenticate_user
from schemas import UserLogin, Message
from typing import List
import uuidapp = FastAPI()# 内存中的连接管理类，用于跟踪和管理活跃的 WebSocket 连接
class ConnectionManager:def __init__(self):# 存储所有活跃的 WebSocket 连接self.active_connections: List[WebSocket] = []# 存储 WebSocket 连接对应的用户名self.users = {}async def connect(self, websocket: WebSocket, username: str):"""添加新的 WebSocket 连接到活跃连接列表，并记录用户名。"""await websocket.accept()  # 接受连接请求self.active_connections.append(websocket)  # 添加到活跃连接列表self.users[websocket] = username  # 记录连接对应的用户名def disconnect(self, websocket: WebSocket):"""从活跃连接列表中移除断开的 WebSocket 连接，并删除对应的用户名记录。"""self.active_connections.remove(websocket)del self.users[websocket]async def broadcast(self, message: str):"""向所有活跃的 WebSocket 连接广播消息。"""for connection in self.active_connections:await connection.send_text(message)async def send_personal_message(self, message: str, websocket: WebSocket):"""向特定的 WebSocket 连接发送消息。"""await websocket.send_text(message)# 创建 ConnectionManager 实例
manager = ConnectionManager()@app.post('/login')
def login(user: UserLogin, Authorize: AuthJWT = Depends()):"""用户登录端点。验证用户凭证，并生成 JWT 令牌。- **username**: 用户名- **password**: 密码"""user_db = authenticate_user(user.username, user.password)  # 验证用户if not user_db:# 如果认证失败，返回 401 错误raise HTTPException(status_code=401, detail="Bad username or password")# 创建 JWT 访问令牌，主题为用户名access_token = Authorize.create_access_token(subject=user.username)return {"access_token": access_token}@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket, Authorize: AuthJWT = Depends()):"""聊天 WebSocket 端点。通过 JWT 令牌认证用户身份，允许用户发送和接收消息。- **token**: 通过查询参数传递的 JWT 令牌"""try:# 从查询参数中获取 JWT 令牌token = websocket.query_params.get("token")if not token:# 如果没有提供令牌，关闭连接并返回未授权错误await websocket.close(code=1008, reason="Unauthorized: No token provided")return# 使用 JWT 进行认证Authorize.jwt_required(auth_from="websocket", token=token)# 获取 JWT 中的主题（用户名）username = Authorize.get_jwt_subject()# 连接并记录用户名await manager.connect(websocket, username)# 向所有连接广播用户加入消息await manager.broadcast(f"{username} joined the chat!")while True:# 接收来自客户端的消息data = await websocket.receive_text()# 广播消息给所有连接的客户端await manager.broadcast(f"{username}: {data}")except WebSocketDisconnect:# 处理客户端断开连接的情况，移除连接并广播离开消息manager.disconnect(websocket)await manager.broadcast(f"{username} left the chat!")except Exception as e:# 处理其他异常，关闭连接并返回内部错误await websocket.close(code=1011, reason="Internal server error")# 可选的简单 HTML 页面用于测试 WebSocket 聊天功能
@app.get("/")
async def get():"""返回一个简单的 HTML 页面，用于测试 WebSocket 聊天功能。"""return HTMLResponse("""<!DOCTYPE html><html><head><title>Chat</title></head><body><h1>WebSocket Chat</h1><!-- 输入 JWT 令牌 --><input id="token" type="text" placeholder="Enter JWT token" /><button οnclick="connect()">Connect</button><br/><br/><!-- 显示聊天记录的文本区域 --><textarea id="chatLog" cols="100" rows="20" readonly></textarea><br/><!-- 输入消息的文本框 --><input id="messageText" type="text" placeholder="Type a message" /><button οnclick="sendMessage()">Send</button><script>var ws;  // WebSocket 连接对象function connect() {var token = document.getElementById("token").value;  // 获取输入的令牌// 创建 WebSocket 连接，传递令牌作为查询参数ws = new WebSocket(`ws://localhost:8000/ws/chat?token=${token}`);// 当收到消息时，追加到聊天记录中ws.onmessage = function(event) {var chatLog = document.getElementById("chatLog");chatLog.value += event.data + "\\n";};// 当连接关闭时，显示关闭原因ws.onclose = function(event) {var chatLog = document.getElementById("chatLog");chatLog.value += "Connection closed: " + event.reason + "\\n";};}function sendMessage() {var input = document.getElementById("messageText");ws.send(input.value);  // 发送消息input.value = '';  // 清空输入框}</script></body></html>""")

4. 运行和测试应用

启动 FastAPI 应用

使用 Uvicorn 启动 FastAPI 应用。

uvicorn main:app --reload

步骤1：用户登录

首先，用户需要通过 /login 端点获取 JWT 令牌。

请求:

curl -X POST "http://localhost:8000/login" \-H "Content-Type: application/json" \-d '{"username": "user1", "password": "password1"}'

响应:

{"access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6..."
}

步骤2：建立 WebSocket 连接

使用获取到的 access_token 建立 WebSocket 连接。

示例连接 URL:

ws://localhost:8000/ws/chat?token=eyJhbGciOiJIUzI1NiIsInR5cCI6...

你可以通过浏览器访问根路径 http://localhost:8000/，在提供的 HTML 页面中输入 JWT 令牌并连接。

步骤3：发送和接收消息

1. 打开浏览器并访问: http://localhost:8000/
2. 输入 JWT 令牌: 将从登录步骤获取的 access_token 粘贴到输入框中。
3. 点击“Connect”按钮: 建立 WebSocket 连接。
4. 发送消息: 在“Type a message”输入框中输入消息并点击“Send”按钮。
5. 查看聊天记录: 所有连接的客户端将接收到广播的消息。

步骤4：断开连接

关闭浏览器或刷新页面，WebSocket 连接将断开，其他连接的客户端将收到离开消息。

5. 处理消息和数据

在上述聊天室应用中，WebSocket 对象用于实时通信，而 Request 对象则用于处理 HTTP 请求，如用户登录。

• 处理 WebSocket 消息:
  • 接收消息: data = await websocket.receive_text()
  • 发送消息: await websocket.send_text(message)
• 处理 HTTP 请求数据:
  • 解析请求体: 通过 Pydantic 模型，如 UserLogin
  • 认证用户: 通过 Authorize 依赖注入和 JWT 令牌生成

6. 测试应用

通过上述步骤，你可以测试聊天室应用的功能：

1. 用户登录: 获取 JWT 令牌。
2. 连接 WebSocket: 使用 JWT 令牌建立连接。
3. 发送消息: 在客户端发送消息，所有连接的用户都会接收到消息。
4. 断开连接: 关闭 WebSocket 连接，其他用户会收到离线通知。

7. 扩展功能

基于这个简单的聊天室，你可以进一步扩展功能：

• 私聊功能: 允许用户发送私密消息。
• 用户列表: 显示当前在线用户。
• 消息持久化: 将消息存储到数据库中，支持聊天历史查询。
• 高级认证: 支持不同权限的用户角色。
• 多房间: 支持多个聊天房间。

---

5. 总结

通过以上系统性的讲解和实战案例，你应该对如何在 FastAPI 中使用 Request 和 WebSocket 对象有了全面的了解。以下是关键点的回顾：

1. Request 对象:
   • 用于处理传统的 HTTP 请求。
   • 提供丰富的方法和属性来访问请求数据。
   • 适用于需要解析请求体、访问查询参数、头信息等场景。
2. WebSocket 对象:
   • 用于实时的双向通信。
   • 允许服务器和客户端之间持续发送和接收消息。
   • 适用于聊天、实时通知、协作工具等场景。
3. 结合使用:
   • 在 WebSocket 连接前，通过 HTTP 请求进行用户认证。
   • 使用 JWT 令牌在 WebSocket 连接中验证用户身份。
   • 管理多个 WebSocket 连接，广播消息或进行一对一通信。
4. 实战案例:
   • 构建了一个简单的聊天室应用，展示了如何结合使用 Request 和 WebSocket 对象。
   • 实现了用户登录、WebSocket 连接、消息广播等功能。
   • 提供了扩展功能的思路，方便根据需求进一步开发。

最佳实践建议

• 模块化设计: 将不同功能的逻辑分离到不同的模块或类中，保持代码的清晰和可维护性。
• 错误处理: 确保 WebSocket 端点具备全面的错误处理机制，能够优雅地处理各种异常情况，避免资源泄漏。
• 安全性: 确保所有 WebSocket 连接都经过认证和授权，防止未授权的访问。
• 性能优化:
  • 使用异步编程模型（如 asyncio）处理高并发的 WebSocket 连接。
  • 合理管理连接池和任务队列，避免阻塞主事件循环。
• 测试覆盖: 编写充分的单元测试和集成测试，确保新功能的稳定性和可靠性。
• 日志记录: 在关键步骤和异常处记录详细的日志，便于调试和监控。