FastAPI（37）- Middleware 中间件

2022-05-15 416

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

本文涉及的产品

服务治理 MSE Sentinel/OpenSergo，Agent数量不受限

简介： FastAPI（37）- Middleware 中间件

什么是中间件

就是一个函数，它在被任何特定路径操作处理之前处理每个请求，且在每个 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