現代 Web 應用越來越重視即時互動和資料傳輸效率。本文將結合 FastAPI、WebSocket 和 GraphQL，展示如何構建一個兼具即時通訊和高效資料查詢能力的 Web 應用。首先，我們會利用 FastAPI 建立 WebSocket 連線管理機制，實作伺服器與多個客戶端之間的雙向通訊和訊息廣播。接著，引入 GraphQL 技術，探討如何定義 Schema、執行查詢和修改操作，以及實作即時資料更新。最後，我們將簡要介紹 FastAPI 的事件處理機制、WSGI 應用掛載，以及安全性考量和測試方法，為開發者提供更全面的技術指導。
WebSocket 多客戶端互動實作與 GraphQL 簡介
WebSocket 多客戶端互動實作
在現代網頁應用程式中，WebSocket 提供了一種高效的雙向通訊機制，使得伺服器能夠主動向客戶端推播訊息。本章節將探討如何使用 FastAPI 實作支援多個 WebSocket 客戶端同時互動的應用程式。
連線管理機制
為了管理多個 WebSocket 連線，我們需要實作一個連線處理器（Connection_handler）。這個類別負責維護當前所有活躍的 WebSocket 連線，並提供廣播訊息給所有連線客戶端的功能。
from typing import List
from fastapi import WebSocket
class Connection_handler:
def __init__(self):
self.connection_list: List[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.connection_list.append(websocket)
def disconnect(self, websocket: WebSocket):
self.connection_list.remove(websocket)
async def send_message(self, message: str, websocket: WebSocket):
await websocket.send_text(message)
async def broadcast(self, message: str):
for connection in self.connection_list:
await connection.send_text(message)
#### 內容解密：
1. `Connection_handler` 類別初始化一個空的 `connection_list` 列表，用於儲存所有活躍的 WebSocket 連線物件。
2. `connect` 方法接受新的 WebSocket 連線請求，並將其新增至 `connection_list`。
3. `disconnect` 方法在客戶端斷開連線時，將對應的 WebSocket 物件從 `connection_list` 中移除。
4. `send_message` 方法向指定的 WebSocket 客戶端傳送訊息。
5. `broadcast` 方法將指定的訊息廣播給 `connection_list` 中的所有客戶端。
#### WebSocket 端點實作
接下來，我們需要在 FastAPI 應用程式中定義一個 WebSocket 端點來處理客戶端的連線請求。
```python
manager = Connection_handler()
@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
await manager.send_message(f"You wrote: {data}", websocket)
await manager.broadcast(f"Client #{client_id} says: {data}")
except WebSocketDisconnect:
manager.disconnect(websocket)
await manager.broadcast(f"Client #{client_id} left the chat")
#### 內容解密：
1. 建立一個 `Connection_handler` 例項 `manager` 來管理所有 WebSocket 連線。
2. 使用 `@app.websocket()` 裝飾器定義 WebSocket 端點 `/ws/{client_id}`，其中 `client_id` 是路徑引數。
3. 當客戶端連線時，呼叫 `manager.connect()` 接受連線並新增至連線列表。
4. 在無限迴圈中接收客戶端傳送的訊息，並將訊息廣播給所有連線的客戶端。
5. 當客戶端斷開連線時，捕捉 `WebSocketDisconnect` 例外，移除斷開的客戶端並廣播通知其他客戶端。
#### 使用者端實作
在使用者端，我們需要建立一個 WebSocket 連線並傳送訊息給伺服器。以下是一個簡單的 HTML 和 JavaScript 實作範例：
```html
<h1>WebSocket Client</h1>
<h2>Your ID: <span id="wsID"></span></h2>
<form action="" onsubmit="sendMessage(event)">
<input type="text" id="sendText"/>
<input type="submit" name="send">
<button onclick="handleOnClick()">Close</button>
</form>
<ul id='messages'>
</ul>
<script>
var client_id = Math.floor(Math.random() * 100);
document.querySelector("#wsID").textContent = client_id;
var ws = new WebSocket(`ws://localhost:8000/ws/${client_id}`);
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) {
event.preventDefault();
var input = document.getElementById("sendText");
ws.send(input.value);
input.value = '';
}
function handleOnClick() {
ws.close();
}
</script>
#### 內容解密：
1. 為每個客戶端產生一個隨機的 `client_id` 並顯示在網頁上。
2. 使用 `WebSocket` API 建立連線到伺服器的 WebSocket 端點。
3. 設定 `onmessage` 事件處理函式來接收伺服器傳送的訊息並顯示在網頁上。
4. 定義 `sendMessage` 函式來傳送使用者輸入的訊息給伺服器。
5. 定義 `handleOnClick` 函式來關閉 WebSocket 連線。
### GraphQL 簡介
GraphQL 是一種用於 API 的查詢語言，它提供了比傳統 RESTful API 更靈活和高效的資料查詢方式。
#### GraphQL 的優勢
1. **精確資料查詢**：客戶端可以明確指定所需的資料欄位，避免過多不必要的資料傳輸。
2. **單一端點**：GraphQL API 通常只暴露一個端點，客戶端透過查詢陳述式來取得所需的資料。
3. **與資料函式庫無關**：GraphQL 不是一種資料函式庫技術，它可以與任何型別的資料來源整合。
#### GraphQL 結構定義語言（SDL）
在 GraphQL 中，我們使用結構定義語言（SDL）來定義 API 的結構。以下是一個簡單的範例：
```graphql
type Book {
title: String!
price: Int!
}
內容解密：
- 定義了一個名為
Book的型別，包含兩個欄位：title和price。 String!和Int!表示這些欄位是非空的，即必須有值。
GraphQL 查詢
客戶端可以透過查詢陳述式來請求所需的資料。以下是一個簡單的查詢範例：
{
books {
title
}
}
內容解密：
- 該查詢請求取得所有書籍的標題。
- 伺服器將根據查詢陳述式傳回相應的資料。
本章節介紹瞭如何使用 FastAPI 實作支援多個 WebSocket 客戶端同時互動的應用程式，並簡要介紹了 GraphQL 的基本概念和優勢。這些技術都可以幫助開發者建立更高效、更靈活的現代網頁應用程式。透過結合 WebSocket 的即時通訊能力和 GraphQL 的精確資料查詢功能，我們可以為使用者提供更好的體驗。未來，這些技術將繼續在網頁開發領域發揮重要作用。
GraphQL 進階應用與實作
GraphQL 查詢與資料擷取
GraphQL 是一種用於 API 的查詢語言，能夠精確指定客戶端所需的資料。查詢（Query）是 GraphQL 中最基本的操作，用於從伺服器擷取資料。以下是一個基本的查詢範例：
{
books {
title
}
}
內容解密：
books是根欄位（root field），代表查詢的起始點。title是books的子欄位，表示要擷取的資料欄位。- GraphQL 的一大優勢是能夠精確指定所需的資料欄位，避免過多不必要的資料傳輸。
查詢的負載（Payload）與資料過濾
在 GraphQL 中，查詢的負載（payload）指的是查詢中指定的欄位。例如，在上述查詢中，title 就是負載的一部分。客戶端可以根據需求動態調整查詢的負載，新增或移除欄位。
{
books {
title
price
}
}
內容解密：
- 這個查詢同時擷取了
title和price兩個欄位。 - GraphQL 的彈性使得客戶端能夠根據實際需求動態調整查詢內容。
Mutations：資料修改操作
除了查詢，GraphQL 還支援 Mutations，用於對資料進行新增、更新或刪除操作。Mutation 的語法結構與查詢類別似，但需要使用 mutation 關鍵字。
mutation {
createBook(title: "Python Programming", price: 750) {
title
price
}
}
內容解密：
createBook是 mutation 的根欄位，用於建立新的書籍資料。title和price是 mutation 的引數，用於指定新書的標題和價格。- mutation 操作傳回新建立的書籍資料，包括
title和price。
Subscriptions：即時資料更新
GraphQL 的 Subscriptions 提供了一種即時資料更新機制，使得客戶端能夠在特定事件發生時立即接收到相關資料。
subscription {
newBook {
title
price
}
}
內容解密：
newBook是 subscription 的根欄位，用於監聽新書建立的事件。- 當伺服器端有新的書籍資料建立時，客戶端會即時接收到相關資料。
- Subscriptions 不遵循傳統的「請求-回應」模式，而是一種持續的資料串流。
GraphQL Schema：定義 API 結構
GraphQL Schema 是 GraphQL API 的核心，定義了客戶端可以進行的操作以及資料的結構。Schema 主要由三種型別組成：Query、Mutation 和 Subscription。
type Query { ... }
type Mutation { ... }
type Subscription { ... }
內容解密：
- Schema 定義了客戶端與伺服器之間的契約。
Query、Mutation和Subscription分別對應了資料查詢、修改和即時更新的操作。
使用 Strawberry 實作 GraphQL API
Strawberry 是 Python 中一個流行的 GraphQL 實作函式庫，與 FastAPI 結合使用可以快速建立高效的 GraphQL API。
首先，安裝 Strawberry：
pip install strawberry-graphql[fastapi]
接著，定義一個簡單的 Book 型別：
import strawberry
@strawberry.type
class Book:
title: str
author: str
price: int
然後，定義 Query 型別並實作一個簡單的查詢：
@strawberry.type
class Query:
@strawberry.field
def book(self) -> Book:
return Book(title="The Godfather", author="Mario Puzo", price=750)
建立 GraphQL Schema 和 FastAPI 應用：
from strawberry.asgi import GraphQL
from fastapi import FastAPI
schema = strawberry.Schema(query=Query)
graphql_app = GraphQL(schema)
app = FastAPI()
app.add_route("/book", graphql_app)
啟動應用後，可以使用 GraphiQL 介面進行查詢操作。
使用 GraphiQL 測試 GraphQL API
GraphiQL 是一個內建於瀏覽器的 GraphQL IDE，能夠方便地測試和除錯 GraphQL API。
query MyQuery {
book {
title
author
price
}
}
執行上述查詢後，可以在輸出視窗中看到結果。
新增 Mutation 操作
除了查詢，Strawberry 也支援 mutation 操作。以下是一個簡單的範例：
@strawberry.type
class Mutation:
@strawberry.mutation
def add_book(self, title: str, author: str, price: int) -> Book:
return Book(title=title, author=author, price=price)
在 GraphiQL 中執行 mutation：
mutation {
addBook(title: "Harry Potter Collection", author: "J.K.Rowling", price:2500) {
title
author
price
}
}
隨著現代應用對即時性和靈活性的需求不斷增加，GraphQL 的優勢將更加明顯。未來，我們可以預見 GraphQL 在以下幾個方面的發展：
- 更好的工具支援：隨著 GraphQL 的普及，將會有更多強大的開發工具和 IDE 支援 GraphQL 的開發和除錯。
- 更廣泛的應用場景：GraphQL 不僅適用於傳統的 Web 應用，還將在行動應用、IoT 等領域發揮重要作用。
- 更強大的生態系統：GraphQL 的生態系統將持續擴充套件，提供更多的函式庫、框架和解決方案，進一步簡化開發流程。
總之，GraphQL 代表著 API 設計的一種新趨勢，具有廣闊的發展前景和巨大的潛力。開發者應當持續關注和學習相關技術，以充分利用 GraphQL 帶來的優勢。
FastAPI 事件處理與安全性探討
FastAPI 提供了多種進階功能，使其成為開發高效能 Web 應用的理想框架。在本章中，我們將探討 FastAPI 的事件處理機制、WSGI 應用的掛載，以及安全性與測試等重要議題。
FastAPI 事件處理
FastAPI 允許開發者針對應用程式的啟動與關閉事件進行自定義處理。這一功能對於需要在應用程式啟動或關閉時執行特定任務的場景非常有用，例如初始化或關閉資料函式庫連線。
from fastapi import FastAPI
import datetime
app = FastAPI()
@app.on_event("startup")
def startup_event():
print('伺服器啟動\n')
log = open("log.txt", mode="a")
log.write("應用程式啟動於 {}\n".format(datetime.datetime.now()))
log.close()
@app.on_event("shutdown")
async def shutdown_event():
print('伺服器關閉 :', datetime.datetime.now())
log = open("log.txt", mode="a")
log.write("應用程式關閉於 {}\n".format(datetime.datetime.now()))
log.close()
內容解密：
- 事件裝飾器：
@app.on_event("startup")和@app.on_event("shutdown")裝飾器分別用於註冊啟動和關閉事件的處理函式。 - 啟動事件處理：在
startup_event函式中，我們開啟一個日誌檔案並記錄應用程式的啟動時間。 - 關閉事件處理：在
shutdown_event函式中，我們同樣記錄應用程式的關閉時間到日誌檔案中。 - 非同步處理：
shutdown_event被定義為非同步函式，以支援非同步操作。
掛載 WSGI 應用程式
FastAPI 支援將其他 WSGI 框架（如 Flask 或 Django）建立的應用程式掛載為子應用程式。這使得開發者能夠在 FastAPI 應用程式中整合現有的 WSGI 應用。
# 主要的 FastAPI 應用程式
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def index():
return {"message": "來自 FastAPI 的 Hello World!"}
# 建立一個 Flask 子應用程式
from flask import Flask
flask_app = Flask(__name__)
@flask_app.route("/")
def index_flask():
return "來自 Flask 的 Hello World!"
# 將 Flask 應用程式掛載到 FastAPI
from fastapi.middleware.wsgi import WSGIMiddleware
app.mount("/flask", WSGIMiddleware(flask_app))
內容解密：
- WSGI 中介軟體：使用
WSGIMiddleware將 Flask 應用包裝，使其能夠被掛載到 FastAPI。 - 掛載路徑：將 Flask 應用掛載到
/flask路徑下，使得兩者能夠共存於同一個伺服器。 - 路由處理：FastAPI 和 Flask 各自處理自己的路由，互不幹擾。
安全性與測試
在開發 API 時，確保只有授權使用者能夠存取特定資源是至關重要的。FastAPI 提供了多種機制來處理安全性問題，包括例外處理和測試。
例外處理
FastAPI 使用 HTTPException 來處理執行階段錯誤，並傳回適當的 HTTP 狀態碼給客戶端。
from fastapi import FastAPI, HTTPException
app = FastAPI()
names = [{"A01": "Alice"}, {"B01": "Bob"}, {"C01": "Christie"}]
@app.get("/names/{id}")
async def get_name(id: str):
for name in names:
if id in name.keys():
return {"name": name[id]}
raise HTTPException(status_code=404, detail="找不到名稱")
內容解密：
- HTTP 例外：當客戶端請求的資源不存在時，
HTTPException被丟擲，並傳回 404 狀態碼。 - 錯誤訊息：提供詳細的錯誤訊息給客戶端，以便其瞭解錯誤原因。
