作者：麦克煎蛋 出处：https://www.cnblogs.com/mazhiyong/ 转载请保留这段声明，谢谢！
我们可以在FastAPI中使用WebSockets建立从前端到后端的通讯连接。
以下我们通过一个简单的网页示例来讲解。
一、创建websocket
在FastAPI应用中创建一个websocket。
然后在WebSocket路由中通过await等待连接和收发消息。这里可以收发文本、二进制以及JSON格式的数据内容。
from fastapi import FastAPI, WebSocket app = FastAPI() @app.websocket("/ws") async def websocket_endpoint(websocket: WebSocket): await websocket.accept() while True: data = await websocket.receive_text() await websocket.send_text(f"Message text was: {data}")
二、建立测试页面
启动时返回默认测试页面。
from fastapi.responses import HTMLResponse html = """ <!DOCTYPE html> <html> <head> <title>Chat</title> </head> <body> <h1>WebSocket Chat</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)
我们可以启动应用并进行收发消息测试：
三、使用功能模块
与路径操作类似，我们可以在WebSocket使用以下功能模块：
DependsSecurityCookieHeaderPathQuery
from typing import Optional from fastapi import Cookie, Depends, FastAPI, Query, WebSocket, status from fastapi.responses import HTMLResponse app = FastAPI() html = """ <!DOCTYPE html> <html> <head> <title>Chat</title> </head> <body> <h1>WebSocket Chat</h1> <form action="" onsubmit="sendMessage(event)"> <label>Item ID: <input type="text" id="itemId" autocomplete="off" value="foo"/></label> <label>Token: <input type="text" id="token" autocomplete="off" value="some-key-token"/></label> <button onclick="connect(event)">Connect</button> <hr> <label>Message: <input type="text" id="messageText" autocomplete="off"/></label> <button>Send</button> </form> <ul id='messages'> </ul> <script> var ws = null; function connect(event) { var itemId = document.getElementById("itemId") var token = document.getElementById("token") ws = new WebSocket("ws://localhost:8000/items/" + itemId.value + "/ws?token=" + token.value); 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) }; event.preventDefault() } 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) 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/{item_id}/ws") async def websocket_endpoint( websocket: WebSocket, item_id: str, 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"Query parameter q is: {q}") await websocket.send_text(f"Message text was: {data}, for item ID: {item_id}")
运行以后可以进行收发消息以及参数测试：
更多信息可以参考以下文档：