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 了

正常传参的请求结果

自定义的请求头和响应码已经生效啦