FastAPI（37）- Middleware 中间件

小菠萝测试笔记

发布于 2021-09-29 16:56:26

2.1K00

代码可运行

文章被收录于专栏：自动化、性能测试自动化、性能测试

运行总次数：0

代码可运行

什么是中间件

就是一个函数，它在被任何特定路径操作处理之前处理每个请求，且在每个 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 或其他功能执行自定义逻辑 ***