求可对接RTK Query流式更新示例的FastAPI WebSocket接口实现案例

• 先安装运行所需依赖：
  pip install fastapi uvicorn pydantic websockets

该实现完全匹配前端逻辑要求：

1. 提供GET /messages/{channel}接口，返回对应频道的历史消息数组，结构和前端Message接口完全对齐
2. 提供WebSocket服务在8080端口，推送的消息格式符合前端校验规则

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Literal, List, Dict, Set

Channel = Literal["redux", "general"]

class Message(BaseModel):
    id: int
    channel: Channel
    userName: str
    text: str

app = FastAPI()

# 配置跨域，允许前端访问，可根据你的前端实际地址调整白名单
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 内存存储调试用数据，生产环境可替换为数据库
# 各频道历史消息
channel_messages: Dict[Channel, List[Message]] = {
    "redux": [
        Message(id=1, channel="redux", userName="system", text="欢迎来到Redux频道"),
        Message(id=2, channel="redux", userName="user1", text="RTK Query Websocket模式适配完成")
    ],
    "general": [
        Message(id=1, channel="general", userName="system", text="欢迎来到通用频道")
    ]
}
# 消息自增ID计数器
id_counter = 3
# 按频道分组存储活跃WebSocket连接
active_connections: Dict[Channel, Set[WebSocket]] = {
    "redux": set(),
    "general": set()
}

# REST接口：获取对应频道历史消息
@app.get("/messages/{channel}", response_model=List[Message])
def get_messages(channel: Channel):
    return channel_messages.get(channel, [])

# WebSocket服务
@app.websocket("/")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    current_channel: Channel | None = None
    try:
        while True:
            # 接收客户端发送的新消息，存储后广播给对应频道所有在线用户
            data = await websocket.receive_json()
            message = Message(**data)
            global id_counter
            # 前端发新消息时可传id=0，由服务端生成自增ID
            if message.id == 0:
                message.id = id_counter
                id_counter += 1
            # 存入历史消息
            channel_messages[message.channel].append(message)
            # 广播给对应频道所有连接
            for conn in active_connections[message.channel]:
                await conn.send_json(message.model_dump())
            # 记录当前连接所属频道，断开时清理
            current_channel = message.channel
            if websocket not in active_connections[current_channel]:
                active_connections[current_channel].add(websocket)
    except WebSocketDisconnect:
        if current_channel and websocket in active_connections[current_channel]:
            active_connections[current_channel].remove(websocket)

使用说明 #

• 启动服务命令：uvicorn main:app --host 0.0.0.0 --port 8080 --reload
• 前端需要调整两处配置：1. fetchBaseQuery的baseUrl改为http://localhost:8080；2. WebSocket连接地址保持ws://localhost:8080即可
• 测试时可通过WebSocket调试工具向服务端发送符合格式的消息，前端RTK Query缓存会自动更新对应频道的消息列表

内容的提问来源于stack exchange，提问作者Mikey