FastAPI 框架以其高效能和易用性著稱，特別適合構建高併發的網路應用。本文將深入探討 FastAPI 的非同步任務處理機制，以及如何結合 JSON 序列化和例外處理機制，打造更強健的 REST API。同時，我們也會探討如何使用 BackgroundTasks 執行非同步操作，例如記錄日誌，以及如何運用 FastAPI 中介軟體實作請求和回應的過濾。此外，文章還會涵蓋 FastAPI 如何與 Asyncio 和 Uvicorn 整合，以提升 API 效能。最後，我們將探討如何使用 FastAPI 的非同步特性來最佳化 API 效能，並示範如何使用 Asyncio 和 Uvicorn 協同工作。
評價系統流程
當使用者提交評價時，系統會進行以下步驟：
- 檢查使用者和旅遊詳細資訊：確保使用者和旅遊的資訊是有效的，如果無效，則丟擲異常。
- 建立評價：生成一個唯一的評價 ID，並建立一個新的評價例項，包含使用者、旅遊和評價等資訊。
- 更新旅遊評分：根據新的評價，更新旅遊的平均評分。
- 新增後臺任務：新增一個後臺任務，記錄使用者的評價操作。
- 傳回評價結果：將評價結果以 JSON 格式傳回給使用者。
程式碼實作
from uuid import uuid1
from fastapi import HTTPException
from fastapi.responses import JSONResponse
from fastapi.encoders import jsonable_encoder
#...
@router.post("/feedback")
def post_tourist_feedback(post: Post):
# 檢查使用者和旅遊詳細資訊
if not post.tourist or not post.tour:
raise HTTPException(status_code=403, detail="使用者和旅遊資訊無效")
# 建立評價
assess_id = uuid1()
assessment = Assessment(id=assess_id, post=post, tour_id=post.tour.id, tourist_id=post.tourist.id)
feedback_tour[assess_id] = assessment
# 更新旅遊評分
tours[post.tour.id].ratings = (tours[post.tour.id].ratings + post.rating) / 2
# 新增後臺任務
bg_task.add_task(log_post_transaction, str(post.tourist.id), message="post_tourist_feedback")
# 傳回評價結果
assess_json = jsonable_encoder(assessment)
return JSONResponse(content=assess_json, status_code=200)
@router.post("/feedback/update/rating")
def update_tour_rating(assess_id: UUID, new_rating: StarRating):
# 檢查評價有效性
if feedback_tour.get(assess_id) is None:
raise HTTPException(status_code=403, detail="評價無效")
# 更新旅遊評分
tour_id = feedback_tour[assess_id].tour_id
tours[tour_id].ratings = (tours[tour_id].ratings + new_rating) / 2
圖表翻譯：
flowchart TD
A[使用者提交評價] --> B[檢查使用者和旅遊資訊]
B --> C[建立評價]
C --> D[更新旅遊評分]
D --> E[新增後臺任務]
E --> F[傳回評價結果]
圖表翻譯：
此圖表展示了評價系統的流程，從使用者提交評價開始，到傳回評價結果為止。每一步驟都對應到程式碼中的實作。
FastAPI 的例外處理機制
FastAPI 提供了一個強大的例外處理機制，允許開發者自訂例外處理函式，以便更好地控制 API 的錯誤回應。在本節中，我們將探討如何使用 FastAPI 的例外處理機制來處理 API 相關的異常。
預設例外處理機制
FastAPI 預設會將異常轉換為 JSON 格式的回應，包含錯誤訊息和狀態碼。然而，開發者可以覆寫這個預設行為，定義自己的例外處理函式。
自訂例外處理函式
開發者可以使用 @app.exception_handler() 裝飾器來定義自訂例外處理函式。例如，以下程式碼定義了一個自訂例外處理函式，將 HTTPException 的回應格式改為 plain text：
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as GlobalStarletteHTTPException
@app.exception_handler(GlobalStarletteHTTPException)
def global_exception_handler(req: Request, ex: str):
return PlainTextResponse(f"Error message: {ex}", status_code=ex.status_code)
處理 RequestValidationError
FastAPI 的 RequestValidationError 例外是由於請求驗證失敗而觸發的。開發者可以定義一個自訂例外處理函式來處理這種例外。例如：
from fastapi.exceptions import RequestValidationError
@app.exception_handler(RequestValidationError)
def validationerror_exception_handler(req: Request, ex: str):
return PlainTextResponse(f"Error message: {str(ex)}", status_code=400)
JSON 相容性
FastAPI 的 JSON 相容性是根據 Python 的 dict、list 和 BaseModel 物件。然而，當處理非 JSON 相容性的物件時，可能會出現異常。開發者可以使用 jsonable_encoder() 函式將物件轉換為 JSON 相容性的 dict 型別。例如：
from fastapi.encoders import jsonable_encoder
class Tourist(BaseModel):
id: UUID
login: User
date_signed: datetime
booked: int
tours: List[TourBasicInfo]
tourist = Tourist(id=uuid1(), login=User(id=uuid1(), username="john"), date_signed=datetime.now(), booked=0, tours=list())
tourist_json = jsonable_encoder(tourist)
在這個例子中，jsonable_encoder() 函式將 Tourist 物件轉換為 JSON 相容性的 dict 型別。
管理 API 回應
在 FastAPI 中，管理 API 回應是一個重要的方面。當處理 API 請求時，需要確保回應是正確的、完整的，並且符合客戶端的需求。在本節中，我們將探討如何使用 JSONResponse 來管理 API 回應。
使用 JSONResponse
JSONResponse 是 FastAPI 中的一個類別，用於傳回 JSON 格式的回應。它可以幫助我們避免 Pydantic 驗證問題，並確保回應是 JSON 友好的。下面是一個例子：
from fastapi import FastAPI, JSONResponse
from fastapi.encoders import jsonable_encoder
app = FastAPI()
@app.post("/signup")
async def signup():
# ...
try:
# ...
except Exception as e:
return JSONResponse(content={"message": "invalid operation"}, status_code=500)
@app.get("/ch02/destinations/details/{id}")
async def check_tour_profile(id: UUID):
tour_info_json = jsonable_encoder(tours[id])
return JSONResponse(content=tour_info_json)
在上面的例子中，JSONResponse 用於傳回 JSON 格式的回應。它可以幫助我們避免 Pydantic 驗證問題，並確保回應是 JSON 友好的。
使用 jsonable_encoder
jsonable_encoder 是 FastAPI 中的一個函式，用於將模型物件轉換為 JSON 格式。它可以幫助我們避免 Pydantic 驗證問題，並確保回應是 JSON 友好的。下面是一個例子：
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
app = FastAPI()
@app.get("/ch02/destinations/list/all")
async def list_tour_destinations():
tours_json = jsonable_encoder(tours)
return JSONResponse(content=tours_json)
在上面的例子中，jsonable_encoder 用於將 tours 模型物件轉換為 JSON 格式。
傳回頭部和 Cookie
JSONResponse 也可以傳回頭部和 Cookie。下面是一個例子：
from fastapi import FastAPI, JSONResponse
app = FastAPI()
@app.get("/ch02/destinations/list/all")
async def list_tour_destinations():
tours_json = jsonable_encoder(tours)
resp_headers = {
'X-Access-Tours': 'Try Us',
'X-Contact-Details': '1-900-888-TOLL',
'Set-Cookie': 'AppName=ITS; Max-Age=3600; Version=1'
}
return JSONResponse(content=tours_json, headers=resp_headers)
在上面的例子中，JSONResponse 傳回了頭部和 Cookie。
背景處理與非同步任務
FastAPI 框架提供了一種機制，允許您在主服務執行的同時，執行背景任務。這些任務可以是任何需要在伺服器執行緒中執行的操作，例如寫入日誌、傳送電子郵件或執行長時間執行的任務。
背景任務的建立
要建立背景任務，您需要使用 BackgroundTasks 類，該類是 FastAPI 框架的一部分。您可以在 API 服務方法的引數列表中宣告它，框架將會注入 BackgroundTask 例項。
以下是建立背景任務的示例：
from fastapi import BackgroundTasks
@router.get("/ch02/tourist/tour/booked")
def show_booked_tours(touristId: UUID, background_tasks: BackgroundTasks):
# ...
background_tasks.add_task(audit_log_transaction, touristId, "booked tour")
# ...
在這個示例中，audit_log_transaction 函式被新增到背景任務中，該函式將寫入日誌檔案。
寫入日誌檔案
以下是 audit_log_transaction 函式的實作：
from datetime import datetime
def audit_log_transaction(touristId: str, message: str):
with open("audit_log.txt", mode="a") as logfile:
content = f"tourist {touristId} executed {message} at {datetime.now()}"
logfile.write(content + "\n")
這個函式將寫入日誌檔案，包含 tourist ID、訊息和當前時間。
HTTP 例外處理
FastAPI 框架還提供了一種機制，允許您處理 HTTP 例外。您可以使用 HTTPException 類來建立自定義的 HTTP 例外。
from fastapi import HTTPException
@router.get("/ch02/tourist/tour/booked")
def show_booked_tours(touristId: UUID):
if approved_users.get(touristId) is None:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="details are missing",
headers={"X-InputError": "missing tourist ID"}
)
# ...
在這個示例中，HTTPException 類被用來建立一個自定義的 HTTP 例外，包含狀態碼、錯誤訊息和頭部資訊。
圖表翻譯：
graph LR
A[API 服務] -->|呼叫|> B[背景任務]
B -->|執行|> C[寫入日誌檔案]
C -->|完成|> D[傳回結果]
D -->|傳回|> A
這個圖表展示了 API 服務、背景任務和寫入日誌檔案之間的關係。
使用 FastAPI 實作背景任務處理
在 FastAPI 中，背景任務（Background Tasks）可以使用 BackgroundTasks 類別來實作。這允許您在主執行緒之外執行任務，從而不阻塞主執行緒。
背景任務的應用場景
背景任務通常用於需要長時間執行的操作，例如傳送郵件、寫入日誌、或是執行複雜的計算任務。在這種情況下，使用背景任務可以提高應用程式的回應速度和使用者經驗。
實作背景任務
以下是使用 FastAPI 實作背景任務的範例：
from fastapi import APIRouter, status, BackgroundTasks
from fastapi.responses import JSONResponse
router = APIRouter()
def audit_log_transaction(touristId: str, message: str):
# 實際的日誌寫入邏輯
print(f"寫入日誌：{touristId} - {message}")
@router.post("/ch02/user/login/")
async def login(login: User, bg_task: BackgroundTasks):
try:
# ...
bg_task.add_task(audit_log_transaction, touristId=str(login.id), message="login")
return JSONResponse(content=signup_json, status_code=status.HTTP_200_OK)
except:
return JSONResponse(content={"message": "invalid operation"}, status_code=status.HTTP_500_INTERNAL_SERVER_ERROR)
@router.get("/ch02/user/login/{username}/{password}")
async def login(username: str, password: str, bg_task: BackgroundTasks):
# ...
bg_task.add_task(audit_log_transaction, touristId=username, message="login")
return JSONResponse(content={"message": "login success"}, status_code=status.HTTP_200_OK)
在這個範例中，audit_log_transaction 函式負責寫入日誌。當使用者登入時，背景任務會被新增到 BackgroundTasks 例項中，從而在主執行緒之外執行日誌寫入任務。
注意事項
- 背景任務不應該用於需要立即傳回結果的操作。
- 背景任務可能會因為主執行緒的結束而被終止，因此需要確保背景任務的執行時間不超過主執行緒的執行時間。
- 背景任務的例外處理需要特別注意，因為背景任務的例外不會被主執行緒捕捉到。
圖表翻譯：
flowchart TD
A[使用者登入] --> B[新增背景任務]
B --> C[執行日誌寫入]
C --> D[傳回登入結果]
圖表翻譯：
此圖表描述了使用者登入的流程。當使用者登入時，系統會新增一個背景任務，負責執行日誌寫入。日誌寫入任務在主執行緒之外執行，從而不阻塞主執行緒。最終，系統會傳回登入結果給使用者。
使用 FastAPI 的非同步路由和背景任務
FastAPI 是一個根據 Python 的非同步網路框架，使用 AsyncIO 原理和概念來建立 REST API 實作。這使得 FastAPI 可以在主執行緒之外執行背景任務，從而提高應用程式的效能。
背景任務
背景任務是指那些不需要立即回應的任務，例如日誌記錄、SMTP/FTP 相關要求、事件和一些資料函式庫相關觸發器。FastAPI 提供了一種簡單的方式來新增背景任務，使用 bg_task 物件。
from fastapi import BackgroundTasks
def audit_log_transaction(touristId: str, message: str):
# 實作日誌記錄邏輯
pass
@router.post("/login")
async def login():
# ...
bg_task.add_task(audit_log_transaction, touristId=str(tourist['login']['id']), message="login")
return JSONResponse(content=tour_json, status_code=status.HTTP_200_OK)
在上面的例子中，audit_log_transaction 函式被新增到背景任務佇列中，當使用者登入時，背景任務會被觸發。
非同步路由
FastAPI 也支援非同步路由，使用 async 關鍵字來定義非同步函式。
@router.get("/feedback/list")
async def show_tourist_post(touristId: UUID):
tourist_posts = [assess for assess in feedback_tour.values() if assess.tourist_id == touristId]
tourist_posts_json = jsonable_encoder(tourist_posts)
return JSONResponse(content=tourist_posts_json, status_code=200)
在上面的例子中，show_tourist_post 函式被定義為非同步函式，使用 async 關鍵字。這使得函式可以在主執行緒之外執行，從而提高應用程式的效能。
非同步 API 端點呼叫非同步函式
非同步 API 端點可以呼叫非同步函式，使用 await 關鍵字。
from utility import check_post_owner
@router.get("/feedback/list")
async def show_tourist_post(touristId: UUID):
# ...
post_owner = await check_post_owner(touristId)
# ...
return JSONResponse(content=tourist_posts_json, status_code=200)
在上面的例子中，check_post_owner 函式被定義為非同步函式，使用 await 關鍵字來呼叫。
使用Asyncio和Uvicorn最佳化REST API效能
在設計REST API時，為了確保非阻塞式的請求處理，使用非同步（async）函式是非常重要的。下面，我們將探討如何使用Asyncio和Uvicorn來最佳化REST API的效能。
Asyncio和Await關鍵字
在Python中，Asyncio是一個用於寫作單執行緒併發程式的函式庫，它可以讓你的程式同時執行多個任務。使用await關鍵字可以讓程式在等待某個任務完成的時候，暫時將控制權交給其他任務。
import asyncio
async def my_function():
# 做一些事情
await asyncio.sleep(1) # 暫停1秒
# 做其他事情
Uvicorn和Worker數量
Uvicorn是一個支援非同步的Web伺服器，它可以讓你的REST API同時處理多個請求。為了提高效能，你可以增加Uvicorn的Worker數量。
uvicorn main:app --workers 5 --reload
這個命令會啟動Uvicorn伺服器，使用5個Worker處理請求。
範例：使用Asyncio和Uvicorn最佳化REST API
下面是一個簡單的範例，示範如何使用Asyncio和Uvicorn最佳化REST API的效能。
from fastapi import FastAPI
import asyncio
app = FastAPI()
async def check_post_owner(feedback_tour, access_id, tourist_id):
# 做一些事情
await asyncio.sleep(1) # 暫停1秒
# 做其他事情
return True
@app.delete("/feedback/delete")
async def delete_tourist_feedback(assess_id: str, tourist_id: str):
if approved_users.get(tourist_id) is None and feedback_tour.get(assess_id):
raise PostFeedbackException(detail='tourist and tour details invalid', status_code=403)
post_delete = [access for access in feedback_tour.values() if access.id == assess_id]
for key in post_delete:
is_owner = await check_post_owner(feedback_tour, access.id, tourist_id)
if is_owner:
del feedback_tour[access.id]
return JSONResponse(content={"message": f"deleted posts of {tourist_id}"}, status_code=200)
在這個範例中，我們使用await關鍵字來等待check_post_owner函式完成，然後再處理請求。同時，我們使用Uvicorn的Worker數量來提高效能。
使用 FastAPI 中介軟體實作請求和回應過濾
FastAPI 中介軟體是一種非同步函式，作為 REST API 服務的過濾器。它可以過濾進入的請求，進行驗證、認證、日誌記錄、背景處理或內容生成等操作。同時，也可以過濾出來的回應，進行內容轉換、回應頭更新和新增等操作。
從系統架構視角來看，本文深入探討了構建高效能 REST API 的關鍵技術，涵蓋評價系統流程、例外處理、API 回應管理、背景任務處理、非同步操作以及效能最佳化等方面。透過多維度分析，我們發現 FastAPI 提供了豐富的功能和工具，例如 BackgroundTasks、JSONResponse、jsonable_encoder 以及非同步路由等，可以有效簡化開發流程，提升 API 效能和開發者體驗。然而，在實際應用中，仍需注意背景任務的資源消耗和錯誤處理，以及非同步操作的複雜性。展望未來，隨著 FastAPI 的不斷發展和社群的壯大，預計會有更多更強大的功能和工具出現，進一步降低開發門檻，提升 API 開發效率。對於追求高效能和可維護性的開發團隊而言，FastAPI 是一個值得深入研究和應用的框架。玄貓認為，FastAPI 的非同步特性和豐富的功能使其在構建高效能 REST API 方面具有顯著優勢，值得開發者關注並應用於實際專案中。
