FastAPI 学习之路（四十四）WebSockets

news/2024/9/13 16:44:04/ 标签: fastapi

我们之前的分析都是基于http的请求，那么如果是websockets可以支持吗，答案是可以的，我们来看下是如何实现的。

from fastapi import WebSocket, FastAPI
from fastapi.responses import HTMLResponseapp = FastAPI()html = """<!DOCTYPE html><html><head><title>Chat</title></head><body><h1>WebSocket 聊天</h1><form action="" onsubmit="sendMessage(event)"><input type="text" id="messageText" autocomplete="off"/><button>Send</button></form><ul id='messages'></ul><script>var ws = new WebSocket("ws://localhost:8000/ws");ws.onmessage = function(event) {var messages = document.getElementById('messages')var message = document.createElement('li')var content = document.createTextNode(event.data)message.appendChild(content)messages.appendChild(message)};function sendMessage(event) {var input = document.getElementById("messageText")ws.send(input.value)input.value = ''event.preventDefault()}
</script></body></html>"""@app.get("/")
async def get_page():return HTMLResponse(html)@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):await websocket.accept()while True:data = await websocket.receive_text()await websocket.send_text(f"接收到的消息是：{data}")

其实很简单，我们在后端写一个接收消息的接口，有消息来了，我们就处理这个消息，不出之前都是处理的http，现在是websocket。我们把拿到的消息直接返回给了前端。

我们看下效果：

如此我们简单的实现了一个websocket的功能。那么我们有一条特定的消息，我们就关闭聊天，如何处理呢？

# 接收特定的消息后，终止该聊天
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):await websocket.accept()while True:data = await websocket.receive_text()if data == "bye":await websocket.send_text(f"请注意：窗口即将关闭")await websocket.send_text("窗口已关闭")await websocket.close()breakelse:await websocket.send_text(f"接收到的消息是：{data}")

前段代码没有改，只是修改了后端。

其实这里去实现websocket通信还是很简单的。

当然我们在请求中也可以使用

Depends
Security
Cookie
Header
Path
Query

我们可以看下面一个简单的后端写的例子

# 使用多种参数，比如获取cookie等
from typing import Optional
from fastapi import Cookie, Query, status, Dependsasync def get_cookie_or_token(websocket: WebSocket,session: Optional[str] = Cookie(None),token: Optional[str] = Query(None),
):if session is None and token is None:await websocket.close(code=status.WS_1008_POLICY_VIOLATION)return session or token@app.websocket("item/ws")
async def websocket_endpoint(websocket: WebSocket,q: Optional[int] = None,cookie_or_token: str = Depends(get_cookie_or_token),
):await websocket.accept()while True:data = await websocket.receive_text()await websocket.send_text(f"Session cookie or query token value is: {cookie_or_token}")if q is not None:await websocket.send_text(f"parameter q is: {q}")await websocket.send_text(f"Message text was: {data}, for : {q}")

虽然我们只是简单的实现了后端，但是可以看出fastapi对于websocket有一个很好的支持