FastAPI WebSocket 即時通訊應用

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 的雙向通訊特性使其非常適合構建即時應用程式。在本文的範例中，我們將擴充套件前面的範例，實作一個支援多個客戶端的聊天應用程式。

  graph LR
    A[客戶端1] -->|WebSocket| B[FastAPI 伺服器]
    C[客戶端2] -->|WebSocket| B
    B -->|廣播訊息|> A
    B -->|廣播訊息|> C

圖表翻譯： 此圖示展示了多個客戶端透過 WebSocket 連線到 FastAPI 伺服器的場景。伺服器接收來自任何客戶端的訊息後，將其廣播給所有已連線的客戶端。

內容解密：

圖表顯示了多個客戶端與伺服器之間的 WebSocket 連線。
伺服器負責接收來自任何客戶端的訊息，並將其廣播給所有連線的客戶端。

本章介紹瞭如何使用 FastAPI 實作 WebSocket 功能，包括基本設定、客戶端實作、測試方法以及一個更複雜的聊天應用範例。透過這些範例，開發者可以更好地理解如何在自己的應用程式中使用 WebSocket 實作即時通訊功能。

玄貓

技術愛好者，專注於分享程式開發、雲端技術與 AI 應用的心得體會。