前言
上一篇我们分享了FastAPI 学习之路（四十三）路径操作的高级配置，这次分享下WebSockets。
正文
我们之前的分析都是基于http的请求，那么如果是WebSockets可以支持吗，答案是可以的，那么我们看下如何实现的。
代码语言：javascript复制from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
app = 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():
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的。我们把拿到的消息直接返回给了前端。
我们可以去测试下。
我们可以看到，我们简单的实现的一个功能就完成了，那么我们现在想着，如果接受到一个特定的消息，我们就关闭说聊天关闭。如何处理呢.
代码语言：javascript复制@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"接受到的消息是: {data}")
await websocket.send_text("聊天关闭")
await websocket.close(100)
else:
await websocket.send_text(f"接受到的消息是: {data}")我们的前端代码没有修改，只是在后端做了下处理。
其实这里去实现websocket通信还是很简单的。
当然我们在请求中也可以使用
DependsSecurityCookieHeaderPathQuery
我们可以看下面一个简单的后端写的例子
代码语言：javascript复制async 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("/items/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}")
这是一个简单的实现，我们只是简单的实现了后端的，前端的其实也简单，就是一个简单的示例的demo。我们可以看到fastapi对于websocket有一个很好的支持。
后记
发现问题，解决问题。遇到问题，慢慢解决问题即可。
欢迎关注雷子说测试开发，后续将会持续为大家分享更多的技术知识
