FastAPI（37）- Middleware 中間件

什么是中間件

• 就是一個函數，它在被任何特定路徑操作處理之前處理每個請求，且在每個 response 返回之前被調用
• 類似鈎子函數

 

執行順序

1. 中間件會接收應用程序中的每個請求 Request
   
2. 針對請求 Request 或其他功能，可以自定義代碼塊
3. 再將請求 Request 傳回路徑操作函數，由應用程序的其余部分繼續處理該請求
4. 路徑操作函數處理完后，中間件會獲取到應用程序生成的響應 Response
5. 中間件可以針對響應 Response 或其他功能，又可以自定義代碼塊
6. 最后返回響應 Response 給客戶端

 

Request

FastAPI 有提供 Request 模塊，但其實就是 starlette 里面的 Request

 

Response

FastAPI 有提供 Response 模塊，但其實就是 starlette 里面的 Response

 

中間件和包含 yield 的依賴項、Background task 的執行順序

1. 依賴項 yield 語句前的代碼塊
2. 中間件
3. 依賴項 yield 語句后的代碼塊
4. 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 或其他功能執行自定義邏輯 ***

 

從請求結果再看執行流程圖

• 黃色塊就是業務代碼啦
• 紅色線就是處理完 Request，准備返回 Response 了

 

正常傳參的請求結果

自定義的請求頭和響應碼已經生效啦