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 连接。
    • 合理管理连接池和任务队列，避免阻塞主事件循环。
  • 测试覆盖: 编写充分的单元测试和集成测试，确保新功能的稳定性和可靠性。
  • 日志记录: 在关键步骤和异常处记录详细的日志，便于调试和监控。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/web/61682.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

深入探索Flax：一个用于构建神经网络的灵活和高效库

深入探索Flax：一个用于构建神经网络的灵活和高效库

深入探索Flax&#xff1a;一个用于构建神经网络的灵活和高效库 在深度学习领域&#xff0c;TensorFlow 和 PyTorch 作为主流的框架&#xff0c;已被广泛使用。不过&#xff0c;Flax 作为一个较新的库&#xff0c;近年来得到了越来越多的关注。Flax 是一个由Google Research团队…
阅读更多...
在1~n中、找出能同时满足用3除余2，用5除余3，用7除余2的所有整数。:JAVA

在1~n中、找出能同时满足用3除余2，用5除余3，用7除余2的所有整数。:JAVA

链接&#xff1a;登录—专业IT笔试面试备考平台_牛客网 来源&#xff1a;牛客网 题目描述 在1~n中、找出能同时满足用3除余2&#xff0c;用5除余3&#xff0c;用7除余2的所有整数。 输入描述: 输入一行&#xff0c;包含一个正整数n &#xff0c;n在100000以内 输出描述:…
阅读更多...
Python 自动化办公的 10 大脚本

Python 自动化办公的 10 大脚本

大家好&#xff0c;我是你们的 Python 讲师&#xff01;今天我们将讨论 10 个实用的 Python 自动化办公脚本。这些脚本可以帮助你简化日常工作&#xff0c;提高效率。无论是处理 Excel 文件、发送邮件&#xff0c;还是自动化网页操作&#xff0c;Python 都能派上用场。 1. 批量…
阅读更多...
UNITY_GOF_ChainOfResponsibility

UNITY_GOF_ChainOfResponsibility

责任链模式 经理和高管可以响应采购请求或将其移交给上级。 每个位置都可以有自己的一套规则&#xff0c;他们可以批准这些订单。 首先有一个领导类专门负责处理这些请求之后根据权限分别对子类进行处理在调用的时候会根据处理类的权限进行处理&#xff0c;如果权限不足就传递给…
阅读更多...
什么是Git

什么是Git

Git Git是什么&#xff1f;Git核心功能Git的常用命令Git原理对象存储Blob对象Tree对象 分支管理版本合并Git的工作流程Git的底层操作 Git是什么&#xff1f; Git是一种分布式版本控制系统&#xff0c;它可以帮助团队协作开发&#xff0c;跟踪代码变更历史&#xff0c;管理和维…
阅读更多...
.net XSSFWorkbook 读取/写入 指定单元格的内容

.net XSSFWorkbook 读取/写入 指定单元格的内容

方法如下&#xff1a; using NPOI.SS.Formula.Functions;using NPOI.SS.UserModel;using OfficeOpenXml.FormulaParsing.Excel.Functions.DateTime;using OfficeOpenXml.FormulaParsing.Excel.Functions.Numeric;/// <summary>/// 读取Excel指定单元格内容/// </summa…
阅读更多...
704. 二分查找 C++

704. 二分查找 C++

文章目录 一、题目链接二、参考代码三、所思所悟 一、题目链接 链接: 704. 二分查找 二、参考代码 int search(const vector<int>& nums, int target) {int left 0; int right nums.size() - 1;//左闭右闭[]while (left < right){int mid (left right) / 2;…
阅读更多...
Matlab搜索路径添加不上

Matlab搜索路径添加不上

发现无论是右键文件夹添加到路径&#xff0c;还是在“设置路径”中专门添加&#xff0c;我的路径始终添加不上&#xff0c;导致代码运行始终报错&#xff0c;后来将路径中的“”加号去掉后&#xff0c;就添加成功了&#xff0c;经过测试&#xff0c;路径中含有中文也可以添加成…
阅读更多...
知乎启用AutoMQ替换Kafka，开辟成本优化与运维提效新纪元

知乎启用AutoMQ替换Kafka，开辟成本优化与运维提效新纪元

作者&#xff1a;知乎在线架构组 王金龙 关于知乎 知乎公司&#xff0c;成立于 2010 年 8 月 10 日&#xff0c;于 2011 年 1 月 26 日正式上线&#xff0c;是中文互联网的高质量问答社区和创作者聚集的原创内容平台。 知乎起步于问答&#xff0c;而超越了问答。知乎以「生…
阅读更多...
Canal mysql数据库同步到es

Canal mysql数据库同步到es

Canal与Elasticsearch集成指南 [可视化工具](https://knife.blog.csdn.net/article/details/126348055)下载Canal 1.1.5版本 (alpha-2) 请从GitHub Releases下载Canal 1.1.5版本中间版本alpha-2。仅需下载canal.deployer和canal.adapter组件。 安装Elasticsearch 确保已安装…
阅读更多...
LeetCode41:缺失的第一个正数

LeetCode41:缺失的第一个正数

原题地址&#xff1a;41. 缺失的第一个正数 - 力扣&#xff08;LeetCode&#xff09; 题目描述 给你一个未排序的整数数组 nums &#xff0c;请你找出其中没有出现的最小的正整数。 请你实现时间复杂度为 O(n) 并且只使用常数级别额外空间的解决方案。 示例 1&#xff1a; 输入…
阅读更多...
Python蒙特卡罗MCMC:优化Metropolis-Hastings采样策略Fisher矩阵计算参数推断应用—模拟与真实数据...

Python蒙特卡罗MCMC:优化Metropolis-Hastings采样策略Fisher矩阵计算参数推断应用—模拟与真实数据...

全文链接&#xff1a;https://tecdat.cn/?p38397 本文介绍了其在过去几年中的最新开发成果&#xff0c;特别阐述了两种有助于提升 Metropolis - Hastings 采样性能的新要素&#xff1a;跳跃因子的自适应算法以及逆 Fisher 矩阵的计算&#xff0c;该逆 Fisher 矩阵可用作提议密…
阅读更多...
相机学习笔记——工业相机的基本参数

相机学习笔记——工业相机的基本参数

0、相机分类 图像颜色不同可以分为黑白相机和彩色相机&#xff1a;相同分辨率下&#xff0c;黑白工业相机相比彩色工业相机精度更高&#xff0c;检测图像边缘时&#xff0c;黑白工业相机成像效果更好。 芯片类型不同可以分为CCD相机和CMOS相机&#xff1a;CCD工业相机具有体积小…
阅读更多...
SpringBoot Web 开发请求参数

SpringBoot Web 开发请求参数

SpringBoot Web 开发请求参数 简单的 web 请求： @RestController public class HelloController {@RequestMapping("sayHello")public String sayHello(){System.out.println("Hello World");return "hello world";} }获取请求参数 简单参数…
阅读更多...
服务器如何隐藏端口才能不被扫描？

服务器如何隐藏端口才能不被扫描？

在服务器上隐藏端口以避免被扫描&#xff0c;是一种增强安全性的措施。虽然完全隐藏端口不太可能&#xff08;因为网络通信本质上需要暴露端口&#xff09;&#xff0c;但可以通过一系列技术手段尽量降低端口被扫描或探测的可能性。以下是详细的实现方法&#xff1a; 1. 更改默…
阅读更多...
Oracle—系统包使用

Oracle—系统包使用

文章目录 系统包dbms_redefinition 系统包 dbms_redefinition 功能介绍&#xff1a;该包体可以实现将Oracle库下的表在线改为分区结构或者重新定义&#xff1b; 说明&#xff1a;在检查表是否可以重定义和开始重定义的过程中&#xff0c;按照表是否存在主键&#xff0c;参数 o…
阅读更多...
专业清洁艺术，还原生活本色——友嘉高效除菌洗碗机

专业清洁艺术，还原生活本色——友嘉高效除菌洗碗机

生活中&#xff0c;每个人都渴望拥有一份洁净的生活环境。而家&#xff0c;作为我们最温馨的港湾&#xff0c;对洁净的追求更是无时无刻不在进行。每当饭后的欢声笑语过后&#xff0c;面对一堆沾满油渍、藏匿着细菌的餐具&#xff0c;我们不禁感到一丝烦忧。然而&#xff0c;有…
阅读更多...
Neo4j 图数据库安装与操作指南(以mac为例)

Neo4j 图数据库安装与操作指南(以mac为例)

目录 一、安装前提条件 1.1 Java环境 1.2 Homebrew&#xff08;可选&#xff09; 二、下载并安装Neo4j 2.1 从官方网站下载 2.1.1 访问Neo4j的官方网站 2.1.2 使用Homebrew安装 三、配置Neo4j 3.1 设置环境变量(可选) 3.2 打开配置文件(bash_profile) 3.2.1 打开终端…
阅读更多...
无人机主控芯片技术与算法详解！

无人机主控芯片技术与算法详解！

一、无人机主控芯片核心技术 高性能CPU&#xff1a; 无人机需要高性能的CPU来处理复杂的飞行控制算法、图像处理和数据传输等任务。目前&#xff0c;无人机的CPU主要有大疆自研的飞控系统、高通提供的无人机设计平台Snapdragon Flight&#xff0c;以及基于开源平台APM、Px4等…
阅读更多...
基于python的汽车数据爬取数据分析与可视化

基于python的汽车数据爬取数据分析与可视化

一、研究背景 基于提供的代码片段和讨论&#xff0c;我们可以得出一个与网络抓取、数据处理和数据可视化相关的研究背景&#xff0c;该背景涉及到汽车行业。以下是研究背景的陈述&#xff1a; "在迅速发展的汽车行业中&#xff0c;准确和及时的数据对各方利益相关者至关…
阅读更多...
最新文章

Copyright @ 2022~2023