我們之前的分析都是基於http的請求，那么如果是WebSockets可以支持嗎，答案是可以的，那么我們看下如何實現的。
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的。我們把拿到的消息直接返回給了前端。
我們可以去測試下。
我們可以看到，我們簡單的實現的一個功能就完成了，那么我們現在想着，如果接受到一個特定的消息，我們就關閉說聊天關閉。如何處理呢.
@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通信還是很簡單的。
當然我們在請求中也可以使用
-
Depends -
Security -
Cookie -
Header -
Path -
Query
我們可以看下面一個簡單的后端寫的例子
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有一個很好的支持。
文章首發在公眾號，歡迎關注。
