FastAPI 提供了簡潔易用的 WebSocket 支援，讓開發者能快速建立即時網路應用。本文除了介紹 WebSocket 的基礎概念和工作原理外，更著重於 FastAPI 的實作細節，包含路由設定、客戶端連線、訊息傳遞等關鍵步驟。同時，文章也示範瞭如何利用 FastAPI 的事件處理機制，在應用程式啟動和關閉時執行特定任務，以及如何結合 GraphQL 強化 API 的彈性與效率，讓讀者能快速掌握 FastAPI 開發 WebSocket 應用的技巧。
進階功能：WebSocket 與 GraphQL
WebSocket 技術解析
WebSocket 是一種根據 TCP 的全雙工通訊協定，能夠在單一連線中實作雙向即時通訊。與傳統的 HTTP 請求/回應模式不同，WebSocket 在建立連線後可持續交換資料，無需重複建立連線。
WebSocket 工作原理
- 連線建立：客戶端透過 WebSocket 握手請求與伺服器建立連線，並包含
Sec-WebSocket-Key標頭。 - 伺服器回應：伺服器確認握手請求，並在
Sec-WebSocket-Auth標頭中傳回金鑰的雜湊值。 - 雙向通訊：連線建立後，伺服器與客戶端可隨時交換訊息，無需遵循 HTTP 協定。
WebSocket 的優勢
- 即時通訊：無需輪詢即可實作即時資料傳輸
- 低延遲：減少了建立連線的開銷
- 全雙工通訊：伺服器與客戶端可同時傳送資料
使用 Python 建置 WebSocket 伺服器
要使用 Python 建置 WebSocket 伺服器，需要安裝 websockets 函式庫。該函式庫根據 Python 的 asyncio 套件，因此需要 Python 3.4 或更高版本。
# 安裝指令
pip install uvicorn[standard]
WebSocket 伺服器範例程式碼
import asyncio
import websockets
async def hello(websocket, path):
name = await websocket.recv()
print(f"收到客戶端訊息：{name}")
greeting = f"你好，{name}！"
await websocket.send(greeting)
print(f"傳送訊息至客戶端：{greeting}")
# 建立 WebSocket 伺服器
async def main():
async with websockets.serve(hello, "localhost", 8765):
print("WebSocket 伺服器已啟動...")
await asyncio.Future() # 無限期執行
# 執行伺服器
asyncio.run(main())
WebSocket 客戶端範例程式碼
import asyncio
import websockets
async def hello():
async with websockets.connect('ws://localhost:8765') as websocket:
name = input("請輸入您的名字：")
await websocket.send(name)
print(f"傳送訊息至伺服器：{name}")
greeting = await websocket.recv()
print(f"收到伺服器訊息：{greeting}")
# 執行客戶端
asyncio.run(hello())
FastAPI 中的 WebSocket 支援
FastAPI 原生支援 WebSocket 協定，使得開發即時網路應用程式變得更加容易。
FastAPI WebSocket 範例
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"收到訊息：{data}")
#### 內容解密：
此範例展示瞭如何在 FastAPI 中建立一個簡單的 WebSocket 端點。客戶端連線到 /ws 路徑後，伺服器會接受連線並進入一個無限迴圈，接收並回傳客戶端傳送的訊息。
@app.websocket("/ws")：定義了一個 WebSocket 端點，路徑為/ws。await websocket.accept()：接受客戶端的 WebSocket 連線請求。await websocket.receive_text()：接收客戶端傳送的文字訊息。await websocket.send_text()：向客戶端傳送文字訊息。
多客戶端聊天應用程式
使用 WebSocket 可以輕鬆實作多客戶端之間的即時通訊。以下是一個簡單的多客戶端聊天應用程式範例：
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
app = FastAPI()
class ConnectionManager:
def __init__(self):
self.active_connections = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
async def send_personal_message(self, message: str, websocket: WebSocket):
await websocket.send_text(message)
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
manager = ConnectionManager()
@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
await manager.broadcast(f"客戶端 {client_id} 說：{data}")
except WebSocketDisconnect:
manager.disconnect(websocket)
#### 內容解密：
此範例實作了一個簡單的多客戶端聊天室。所有連線到 /ws/{client_id} 的客戶端都會被加入到活躍連線列表中，當任何客戶端傳送訊息時，伺服器會將該訊息廣播給所有其他客戶端。
ConnectionManager類別：負責管理所有活躍的 WebSocket 連線，並提供連線、斷開連線、傳送個人訊息和廣播訊息的方法。websocket_endpoint路徑操作函式：處理客戶端的 WebSocket 連線請求，並在連線建立後進入訊息接收迴圈，將接收到的訊息廣播給所有其他客戶端。
GraphQL 與 FastAPI 事件
除了 WebSocket，FastAPI 還支援 GraphQL 和事件處理等進階功能，這些功能使得建構現代化網路應用程式變得更加靈活和強大。
GraphQL 簡介
GraphQL 是一種用於 API 的查詢語言，它允許客戶端明確指定所需的資料結構，從而減少了資料傳輸量並提高了 API 的靈活性。
FastAPI 事件處理
FastAPI 提供了事件處理機制，允許開發者在應用程式啟動或關閉時執行特定的程式碼，例如初始化資源或清理資源。
from fastapi import FastAPI
app = FastAPI()
@app.on_event("startup")
async def startup_event():
print("應用程式啟動中...")
@app.on_event("shutdown")
def shutdown_event():
print("應用程式關閉中...")
#### 內容解密：
此範例展示瞭如何在 FastAPI 中使用事件處理機制。在應用程式啟動時，會執行 startup_event 函式，而在應用程式關閉時，會執行 shutdown_event 函式。
@app.on_event("startup")：註冊一個在應用程式啟動時執行的事件處理函式。@app.on_event("shutdown")：註冊一個在應用程式關閉時執行的事件處理函式。
使用 FastAPI 實作 WebSocket 通訊
WebSocket 是一種雙向通訊協定，允許伺服器和客戶端之間進行即時資料交換。在本章中，我們將探討如何使用 FastAPI 框架實作 WebSocket 功能，包括基本設定、客戶端實作以及測試方法。
WebSocket 基本設定
首先，我們需要在 FastAPI 應用程式中定義一個 WebSocket 路由。這可以透過使用 @app.websocket() 裝飾器來完成。下面是一個簡單的例子：
from fastapi import FastAPI, WebSocket
app = FastAPI()
@app.websocket("/test")
async def test(websocket: WebSocket):
await websocket.accept()
while True:
request = await websocket.receive_text()
print(request)
i = random.randint(1, 1000)
await websocket.send_text(str(i))
if i == 100:
break
內容解密：
- 首先，我們匯入必要的模組並建立一個 FastAPI 應使用案例項。
- 使用
@app.websocket("/test")定義了一個 WebSocket 路由，當客戶端連線到/test端點時，將呼叫test函式。 await websocket.accept()用於接受客戶端的連線請求。- 在無限迴圈中，伺服器接收客戶端傳送的文字訊息，並列印到控制檯。
- 伺服器隨機生成一個介於 1 和 1000 之間的整數，並將其作為文字訊息發送回客戶端。
- 如果生成的隨機數為 100，則迴圈終止，連線保持開啟狀態，直到客戶端斷開連線。
客戶端實作
客戶端可以使用 JavaScript 在網頁中實作 WebSocket 連線。以下是一個簡單的客戶端範例：
<script>
var ws = new WebSocket("ws://localhost:8000/test")
ws.onmessage = event => {
var number = document.getElementById("number")
number.innerHTML = event.data
}
handleOnClick = () => {
ws.send("Hi WebSocket Server")
}
</script>
<h3>Streaming numbers appear here</h3>
<div id="number"></div>
<button onclick="handleOnClick()">Click Me</button>
內容解密：
- 使用
new WebSocket("ws://localhost:8000/test")建立與伺服器的 WebSocket 連線。 ws.onmessage事件處理函式用於接收伺服器傳送的訊息，並更新頁面上的元素內容。- 當使用者點選按鈕時，
handleOnClick函式被呼叫，向伺服器傳送一條訊息。
更複雜的聊天應用範例
下面是一個更複雜的範例，展示了一個簡單的聊天應用程式：
伺服器端程式碼：
from fastapi import FastAPI, WebSocket, HTMLResponse
app = FastAPI()
@app.get("/", response_class=HTMLResponse)
async def hello(request: Request):
file = open("templates/socket.html")
html = file.read()
return HTMLResponse(content=html)
@app.websocket("/ws")
async def ws_handler(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Message text was: {data}")
客戶端 JavaScript 程式碼：
function sendMessage(event) {
var input = document.getElementById("sendText")
ws.send(input.value)
input.value = ''
event.preventDefault()
}
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)
};
內容解密：
- 伺服器端程式碼接受客戶端的 WebSocket 連線，並在接收到訊息後回顯該訊息。
- 客戶端 JavaScript 程式碼負責傳送使用者輸入的訊息，並將伺服器回顯的訊息新增到頁面上的列表中。
使用 Insomnia 測試 WebSocket
除了使用前端應用程式外，還可以使用像 Insomnia 這樣的工具來測試 WebSocket API 端點。以下是測試步驟：
- 下載並安裝 Insomnia。
- 建立一個新的 WebSocket 請求。
- 輸入 URL
http://localhost:8000/ws並點選連線按鈕。 - 傳送訊息並觀察伺服器的回應。
多客戶端聊天應用程式
WebSocket 的雙向通訊特性使其非常適合構建即時應用程式。在本文的範例中，我們將擴充套件前面的範例，實作一個支援多個客戶端的聊天應用程式。
@startuml
skinparam backgroundColor #FEFEFE
skinparam defaultTextAlignment center
skinparam rectangleBackgroundColor #F5F5F5
skinparam rectangleBorderColor #333333
skinparam arrowColor #333333
title 多客戶端聊天應用程式
rectangle "WebSocket" as node1
rectangle "廣播訊息" as node2
node1 --> node2
@enduml
圖表翻譯： 此圖示展示了多個客戶端透過 WebSocket 連線到 FastAPI 伺服器的場景。伺服器接收來自任何客戶端的訊息後，將其廣播給所有已連線的客戶端。
內容解密：
- 圖表顯示了多個客戶端與伺服器之間的 WebSocket 連線。
- 伺服器負責接收來自任何客戶端的訊息，並將其廣播給所有連線的客戶端。
本章介紹瞭如何使用 FastAPI 實作 WebSocket 功能，包括基本設定、客戶端實作、測試方法以及一個更複雜的聊天應用範例。透過這些範例，開發者可以更好地理解如何在自己的應用程式中使用 WebSocket 實作即時通訊功能。