FastAPI（37）- Middleware 中介軟體

阿新 • • 發佈：2021-09-27

什麼是中介軟體

就是一個函式，它在被任何特定路徑操作處理之前處理每個請求，且在每個 response 返回之前被呼叫
類似鉤子函式

執行順序

中介軟體會接收應用程式中的每個請求 Request
針對請求 Request 或其他功能，可以自定義程式碼塊
再將請求 Request 傳回路徑操作函式，由應用程式的其餘部分繼續處理該請求
路徑操作函式處理完後，中介軟體會獲取到應用程式生成的響應 Response
中介軟體可以針對響應 Response 或其他功能，又可以自定義程式碼塊
最後返回響應 Response 給客戶端

Request

FastAPI 有提供 Request 模組，但其實就是 starlette 裡面的 Request

Response

FastAPI 有提供 Response 模組，但其實就是 starlette 裡面的 Response

中介軟體和包含 yield 的依賴項、Background task 的執行順序

依賴項 yield 語句前的程式碼塊
中介軟體
依賴項 yield 語句後的程式碼塊
Background task

建立中介軟體

import time
from fastapi import FastAPI, Request


@app.middleware("http")
# 必須用 async
async def add_process_time_header(request: Request, call_next):
    start_time  
= time.time()
    # 必須用 await
    response = await call_next(request)
    process_time = time.time() - start_time
    # 自定義請求頭
    response.headers["X-Process-Time"] = str(process_time)
    # 返回響應
    return response

中介軟體函式接收兩個引數

request：Request 請求，其實就是 starlette 庫裡面的 Request
call_next：是一個函式，將 request 作為引數

call_next

會將 request 傳遞給相應的路徑操作函式
然後會返回路徑操作函式產生的響應，賦值給 response
可以在中介軟體 return 前對 response 進行操作

實際栗子

import uvicorn
from fastapi import FastAPI, Request, Query, Body, status
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

app = FastAPI()


@app.middleware("http")
# 必須用 async
async def add_process_time_header(request: Request, call_next):
    # 1、可針對 Request 或其他功能，自定義程式碼塊
    print("=== 針對 request 或其他功能執行自定義邏輯程式碼塊 ===")
    print(request.query_params)
    print(request.method)

    # 2、將 Request 傳回給對應的路徑操作函式繼續處理請求
    # 必須用 await
    response = await call_next(request)
    # 4、接收到路徑操作函式所產生的的 Response，記住這並不是返回值（return）

    # 5、可針對 Response 或其他功能，自定義程式碼塊
    print("*** 針對 response 或其他功能執行自定義邏輯 ***")

    # 自定義請求頭響應狀態碼
    response.headers["X-Process-Token"] = str("test_token_polo")
    response.status_code = status.HTTP_202_ACCEPTED

    # 6、最終返回 Response 給客戶端
    return response


class User(BaseModel):
    name: str = None
    age: int = None


@app.post("/items/")
async def read_item(item_id: str = Query(...), user: User = Body(...)):
    # 3、收到請求，處理請求
    res = {"item_id": item_id}
    if user:
        res.update(jsonable_encoder(user))
    print("@@@ 執行路徑操作函式 @@@", res)
    
    # 有沒有 return 都不影響中介軟體接收 Response
    return res

重點

call_next 是一個函式，呼叫的就是請求路徑對應的路徑操作函式
返回值是一個 Response 型別的物件

訪問 /items ，控制檯輸出結果

=== 針對 request 或其他功能執行自定義邏輯程式碼塊 ===
item_id=test
POST
@@@ 執行路徑操作函式 @@@ {'item_id': 'test', 'name': 'string', 'age': 0}
*** 針對 response 或其他功能執行自定義邏輯 ***