Python框架篇(5):FastApi-中间件使用

1.介绍

1.1 官网介绍

"中间件"是一个函数,它在每个请求被特定的路径操作处理之前,以及在每个响应返回之前工作.

  • 它接收你的应用程序的每一个 请求.
  • 然后它可以对这个 请求做一些事情或者执行任何需要的代码.
  • 然后它将 请求传递给应用程序的其他部分 (通过某种 路径操作).
  • 然后它获取应用程序生产的 响应 (通过某种 路径操作).
  • 它可以对该 响应做些什么或者执行任何需要的代码.
  • 然后它返回这个 响应.

1.2 中间件工作示意图

alt

1.3 官方使用示例

import time
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

2. 快速使用

从官方示例可以看出，中间件函数要和FastAPI实例在一个文件才能通过注解的方式，这种虽然使用起来比较简单，但是不太合适扩展和项目结构管理，下面是通过函数add_middleware来注册中间件。

2.1 创建中间件

在包app/middleware下，并新增文件usetime_middleware.py，文件内容如下:

import time
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response


class UseTimeMiddleware(BaseHTTPMiddleware):
    """ 计算耗时中间件"""

    def __init__(self, app):
        super().__init__(app)

    async def dispatch(self, request: Request, call_next) -> Response:
        """ 请求耗时 """
        start_time = time.time()
        # 调用下一个中间件或路由处理函数
        result = await call_next(request)
        process_time = time.time() - start_time
        result.headers["X-Process-Time"] = str(process_time)
        return result

@注意:我们定义的中间件类UseTimeMiddleware要继承基础类BaseHTTPMiddleware

2.2 封装注册函数

在包app/middleware/__init__.py引用,并封装统一注册方法：

from fastapi import FastAPI
from .usetime_middleware import UseTimeMiddleware

def registerMiddlewareHandle(server: FastAPI):
    # 添加耗时请求中间件
    server.add_middleware(UseTimeMiddleware)

2.3 调用注册函数

修改main.py文件，修改内容如下:

from app import errors, middleware
...

# 实例化
server = FastAPI(redoc_url=None, docs_url="/apidoc", title="FastAPI学习")

# # 注册中间件
middleware.registerMiddlewareHandle(server)
...

2.4 添加路由

修改app/router/demo_router.py文件，新增内容如下:

@router.get("/middle/useTime")
async def middleUseTime() -> response.HttpResponse:
    """
    中间件使用-演示
    """
    # 随机暂停时间
    seconds = random.randint(500, 5000) / 1000
    print("暂停时间:", seconds)
    time.sleep(seconds)
    return response.ResponseSuccess(seconds)

2.5 验证

alt

3.多中间件顺序

3.1 创建多个中间件

# 文件: app/middleware/test_middleware.py
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response
class TestMiddleware(BaseHTTPMiddleware):
    """ 测试顺序中间件"""

    def __init__(self, app):
        super().__init__(app)

    async def dispatch(self, request: Request, call_next) -> Response:
        print("调用-中间件-TestMiddleware---before")
        # 调用下一个中间件或路由处理函数
        result = await call_next(request)
        print("调用-中间件-TestMiddleware---after")
        return result
 

# -------------- 另外一个中间件  --------------

# 文件: app/middleware/test_middleware.py
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response


class TokenMiddleware(BaseHTTPMiddleware):
    """ token验证中间件 """

    def __init__(self, app):
        super().__init__(app)

    async def dispatch(self, request: Request, call_next) -> Response:
        token = request.headers.get("X-Token", "")
        print("调用-token验证中间件-TokenMiddleware---before", token)
        result = await call_next(request)
        print("调用-token验证中间件-TokenMiddleware---after", token)
        return result

3.2 注册

修文件app/middleware/__init__.py中，统一注册中间件方法：

from fastapi import FastAPI
from .usetime_middleware import UseTimeMiddleware
from .token_middleware import TokenMiddleware
from .test_middleware import TestMiddleware


def registerMiddlewareHandle(server: FastAPI):
    # 添加token验证中间件
    server.add_middleware(TokenMiddleware)
    # 添加耗时请求中间件
    server.add_middleware(UseTimeMiddleware)
    # 测试
    server.add_middleware(TestMiddleware)

3.3 请求验证

alt

@总结: 通过实践发现: 中间件的执行顺序和注册顺序，正好是相反的；先注册的后执行，

3.4 add_middleware函数

    def add_middleware(self, middleware_class: type, **options: typing.Any) -> None:
        if self.middleware_stack is not None:  # pragma: no cover
            raise RuntimeError("Cannot add middleware after an application has started")
         # 发现每次调用，都会把最后一个中间件放在 索引=0的位置，即最前面
        self.user_middleware.insert(0, Middleware(middleware_class, **options))

3.5 优化注册函数

add_middleware函数通过栈的方式(后进先出)注册中间件，跟我们的思维相反，如果注册的中间件多的话，排查问题时，很容易被惯性思维误导，下面优化了下注册函数: 先注册的先执行

from fastapi import FastAPI
from .usetime_middleware import UseTimeMiddleware
from .token_middleware import TokenMiddleware
from .test_middleware import TestMiddleware

# 定义注册顺序
middlewareList = [
    UseTimeMiddleware,  # 添加耗时请求中间件
    TokenMiddleware,  # 添加token验证中间件
    TestMiddleware  # 测试中间件
]


def registerMiddlewareHandle(server: FastAPI):
    # 倒序中间件
    middlewareList.reverse()
    # 遍历注册
    for _middleware in middlewareList:
        server.add_middleware(_middleware)

4.内置中间件

4.1 常用中间件

框架也提供了一些常用的的内置中间件

  • HTTPSRedirectMiddleware: 将 HTTP 请求重定向到 HTTPS。这个中间件会检查请求的协议，如果是 HTTP，则自动将请求重定向到相应的 HTTPS 地址;
  • TrustedHostMiddleware: 强制所有传入请求都具有正确设置的 Host 标头，以防止 HTTP 主机标头攻击。
  • GZipMiddleware: 用于在响应中压缩内容，以减小传输大小。这有助于提高应用程序的性能，特别是在处理大量文本或数据时。
  • CORSMiddleware: 用于处理跨域资源共享（CORS）请求。CORS 是一种浏览器机制，允许 Web 页面从不同的域请求不同域的资源。

4.2 CORSMiddleware使用

跨域中间件应该是我们常用的一种中间件，具体使用示例如下:

# 导入
from fastapi.middleware.cors import CORSMiddleware

# 注册
 server.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],  # 允许的来源，可以是字符串、字符串列表，或通配符 "*"
        allow_credentials=True,  # 是否允许携带凭证（例如，使用 HTTP 认证、Cookie 等）
        allow_methods=["*"],  # 允许的 HTTP 方法，可以是字符串、字符串列表，或通配符 "*"
        allow_headers=["*"],  # 允许的 HTTP 头信息，可以是字符串、字符串列表，或通配符 "*"
        expose_headers=["*"],  # 允许前端访问的额外响应头，可以是字符串、字符串列表
        max_age=600,  # 请求的缓存时间，以秒为单位
    )

以下是常用参数的详细说明：

  • allow_origins: 允许的来源。可以是字符串、字符串列表，或通配符 "*" 表示允许所有来源。
  • allow_credentials: 是否允许携带凭证（例如，使用 HTTP 认证、 Cookie 等）。默认为 False,如果为 True， allow_origins 必须为具体的源，不可以是 ["*"]。
  • allow_methods: 允许的 HTTP 方法，可以是字符串、字符串列表，或通配符 "*" 表示允许所有方法。
  • allow_headers: 允许的 HTTP 头信息，可以是字符串、字符串列表，或通配符 "*" 表示允许所有头信息。
  • expose_headers: 允许前端访问的额外响应头，可以是字符串、字符串列表,一般很少指定。
  • max_age: 浏览器缓存 CORS 响应的最长时间，单位是秒。默认为 600,一般使用默认值。

4.3 GZipMiddleware使用

# 导入
from fastapi.middleware.gzip import GZipMiddleware

# 注册
server.add_middleware(
    GZipMiddleware,
    minimum_size=500,   # 启用 Gzip 压缩的最小响应体大小，单位为字节
    compress_level=6,   # Gzip 压缩级别，范围为 0 到 9，级别越高，压缩率越高，但耗费的 CPU 越多
    exclude_mediatypes=["application/json"],  # 不进行 Gzip 压缩的媒体类型，可以是字符串或字符串列表
)

以下是常用参数的详细说明：

  • minimum_size: 启用 Gzip 压缩的最小响应体大小，单位为字节。只有当响应体的内容大于等于 minimum_size 时，才会进行 Gzip 压缩。默认为 500 字节。
  • compress_level: Gzip 压缩级别，范围为 0 到 9。级别越高，压缩率越高，但耗费的 CPU 越多。默认为 6。
  • exclude_mediatypes: 不进行 Gzip 压缩的媒体类型，可以是字符串或字符串列表。例如，可以配置不对 JSON 响应进行压缩。

5.JWT验证

FastAPI框架没有内置的 JWT（JSON Web Token）中间件，我们可以使用第三方库（ PyJWT）来验证和解码 JWT，下面来实现使用示例；

5.1 安装

$ pip install pyjwt

5.2 生成token

# 定义秘钥和算法
secret_key = "abcd12345@abcdef"
algorithm = "HS256"

def jwtGenerator() -> str:
    # 构造 payload
    payload = {
        "uid": 1234567,  # 主题，通常是用户的唯一标识
        "iat": datetime.utcnow(),  # 签发时间
        "exp": datetime.utcnow() + timedelta(minutes=30),  # 过期时间
        "data": {"user_name": "张三", "uid": 1234567, "phone": "17600000000"}  # 自定义的数据
    }
    # 生成 JWT
    return jwt.encode(payload, secret_key, algorithm=algorithm)
 

 if __name__ == '__main__':
    # 生成token
    token = jwtGenerator()
    print("Token:", token)
 

"""
Token: eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1aWQiOjEyMzQ1NjcsImlhdCI6MTcwMjUyODA1MSwiZXhwIjoxNzAyNTI5ODUxLCJkYXRhIjp7InVzZXJfbmFtZSI6Ilx1NWYyMFx1NGUwOSIsInVpZCI6MTIzNDU2NywicGhvbmUiOiIxNzYwMDAwMDAwMCJ9fQ.sV_k75YPdEtI2P7-HlDbGbXWdcdosf1cCImNkux7OGg
"""

5.3 校验token

def parseToken(jwtToken: str) -> str:
    """ 解析token """
    try:
        return jwt.decode(jwtToken, secret_key, algorithms=[algorithm])
    except jwt.ExpiredSignatureError:
        print("JWT has expired.")
        return ""
    except jwt.InvalidTokenError:
        print("Invalid JWT.")
        return ""
  
 
 if __name__ == '__main__':
    token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1aWQiOjEyMzQ1NjcsImlhdCI6MTcwMjUyODA1MSwiZXhwIjoxNzAyNTI5ODUxLCJkYXRhIjp7InVzZXJfbmFtZSI6Ilx1NWYyMFx1NGUwOSIsInVpZCI6MTIzNDU2NywicGhvbmUiOiIxNzYwMDAwMDAwMCJ9fQ.sV_k75YPdEtI2P7-HlDbGbXWdcdosf1cCImNkux7OGg"
    # 验证token
    parseToken = parseToken(token)
    print("parseToken:", parseToken)
    
"""
parseToken: {'uid': 1234567, 'iat': 1702528051, 'exp': 1702529851, 'data': {'user_name': '张三', 'uid': 1234567, 'phone': '17600000000'}}
"""    

5.4 封装中间件

在包app/middleware下，并新增文件jwt_middleware.py，文件内容如下:

from fastapi import Request, status
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from starlette.responses import Response

from app.types import response, JwtData
from app.utils import JwtManageUtil

# 后期做配置，这里临时演示
secret_key = "abcd12345@abcdef"

# 不检查
noCheckTokenPathList = [
    "/apidoc",
    "/openapi.json",
    "/api/user/login"
]


class JwtMiddleware(BaseHTTPMiddleware):
    """ jwt验证中间件 """

    def __init__(self, app):
        super().__init__(app)
        self.jwtUtil = JwtManageUtil(secretKey=secret_key)

    async def dispatch(self, request: Request, call_next):
        # 判断路由是否需要验证
        path = request.url.path
        if path in noCheckTokenPathList:
            return await call_next(request)
        # 获取token
        token = request.headers.get('x-token', '')
        if token == "":
            return JSONResponse(
                status_code=status.HTTP_200_OK,
                content=jsonable_encoder(response.ResponseFail('token不能为空~')))

        # 验证token
        tokenInfo = self.jwtUtil.decode(token, JwtData)
        if not isinstance(tokenInfo, JwtData):
            # 验证失败
            return JSONResponse(
                status_code=status.HTTP_200_OK,
                content=jsonable_encoder(response.ResponseFail(tokenInfo)))

        result = await call_next(request)
        print("token解析成功", tokenInfo)
        return result

限于文章篇幅，上面示例中JwtManageUtil代码不在展示，具体源代码可在 微信搜索【猿码记】回复 【fastapi】获取

5.5 请求验证

$ curl -X 'GET' \
  'http://127.0.0.1:8000/demo/middle/useTime' \
  -H 'accept: application/json' \
  -H 'X-Token: eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJqdGkiOiIyMDIzMTIxNDIwMzEyNzk2ODUiLCJpc3MiOiJcdTczM2ZcdTc4MDFcdThiYjAiLCJpYXQiOjE3MDI1NTcxMDYsImV4cCI6MTcwMjU2MDcwNiwiZGF0YSI6eyJ1aWQiOjExMjIzMzQ0LCJ1bmFtZSI6Ilx1NWYyMFx1NGUwOSJ9fQ.62Nijbs08Oy1wo1-IBfO9RvTGQ3B6aGtaAQt0qrT7-4'
# 返回
{"code":200,"msg":"处理成功","data":1.096,"additional":{"time":"2023-12-14 20:35:00","trace_id":"90060a47fa9850d1ccb1bc9fc1045c8c"}}

# 故意写错token
$ curl -X 'GET' \
  'http://127.0.0.1:8000/demo/middle/useTime' \
  -H 'accept: application/json' \
  -H 'X-Token: 12334'
# 返回
{"code":-1,"msg":"TokenInvalid|token非法","data":null,"additional":{"time":"2023-12-14 20:35:26","trace_id":"5303bc01350d7ab7f029098378d6a264"}}

本文由 mdnice 多平台发布

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：/a/253155.html

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

全球汽车行业的数字化转型：产品和后端的渐进之旅

全球汽车行业的数字化转型：产品和后端的渐进之旅

如何管理汽车行业的数字化转型?在我们本篇文章中了解更多有关如何设定长期目标的信息。 正在改变汽车行业的26个数字化主题 最近一篇关于汽车行业数字化转型的论文确定了26个数字技术主题（论文详情请点击阅读原文），分为三个主要集群: 1)驾驶…
阅读更多...
添加E1000网卡进行测试，只有VMXNET3性能的四分之一

添加E1000网卡进行测试，只有VMXNET3性能的四分之一

正文共：1444 字 14 图，预估阅读时间：2 分钟 我们前面介绍了VMware ESXi 6.7中的适配器类型性能（VMWare ESXi中，不同的虚拟网卡性能竟然能相差三倍！），当时的配置项主要为E1000e和VMXN…
阅读更多...
【LangChain学习之旅】—（3） LangChain快速构建本地知识库的智能问答系统

【LangChain学习之旅】—（3） LangChain快速构建本地知识库的智能问答系统

【LangChain学习之旅】—（3） LangChain快速构建本地知识库的智能问答系统 项目及实现框架开发框架核心实现机制数据准备及加载加载文本文本的分割向量数据库存储文本的“嵌入”概念向量数据库概念 相关信息获取RetrievalQA生成回答并展示示例小结 Refere…
阅读更多...
MathBuddyGUI：MATLAB多功能计算器，2060行代码

MathBuddyGUI：MATLAB多功能计算器，2060行代码

是我做的一个MATLAB课设，是一个带画图、输出模式转换、简单控制系统仿真等功能的计算器。练习GUI编程用。 仓库链接： MathBuddyGUI: MATLAB课设，一个带画图、输出模式转换、简单控制系统仿真等功能的计算器，练习GUI编程用。 (gi…
阅读更多...
C++——C++11（1）

C++——C++11（1）

时至今日，C标准已经到了C23，但是你要说哪一次提出的标准最经 典，那C11一定会被人提及，C11带来了数量可观的变化，其中包 含了约140个新特性，以及对C03标准中约600个缺陷的修正，这使得 C11更像是从…
阅读更多...
在用 App 设计工具创建的 App 内共享数据

在用 App 设计工具创建的 App 内共享数据

目录 定义属性 访问属性 示例 共享绘图数据和下拉列表选择 使用属性是在 App 内共享数据的最佳方法，因为属性可供 App 内的所有函数和回调访问。所有 UI 组件都是属性，因此可以使用以下语法来访问和更新回调中的 UI 组件： app.Component…
阅读更多...
Java 第12章 异常 本章作业

Java 第12章 异常 本章作业

1 编程 两数相除的异常处理 各自属于哪些异常： 数据格式不正确 NumberformatException 缺少命令行参数 ArrayIndexOutOfBoundsException 除0异常处理 ArithmeticException ArrayIndexOutOfBoundsException 为数组下标越界时会抛出的异常，可以在检测到命…
阅读更多...
计算机网络知识点

计算机网络知识点

计算机网络中的OSI模型 OSI模型是指“国际标准化组织(SO)”提出的使各种计算机在世界范围内互通互联的网络标准框架简称开放系统互联参考模型 (OSI)。 七层模型：应用层、表示层、会话层、传输层、网络层（IP协议、RARP协议、ARP协议、CIDR协议&#xff0…
阅读更多...
0x31 质数

0x31 质数

0x31 质数 定义： 若一个正整数无法被除了1和它自身之外的任何自然数整除，则称该数为质数（或素数），否则则称该正整数为合数。 在整个自然数集合中，质数的数量不多，分布比较稀疏，对…
阅读更多...
机器学习项目精选 第一期：超完整数据科学资料合集

机器学习项目精选 第一期：超完整数据科学资料合集

大噶吼，不说废话，分享一波我最近看过并觉得非常硬核的资源，包括Python、机器学习、深度学习、大模型等等。 1、超完整数据科学资料合集 地址：https://github.com/krishnaik06/The-Grand-Complete-Data-Science-Materials Pytho…
阅读更多...
strlen的三种模拟实现方法

strlen的三种模拟实现方法

首先&#xff0c;我们要了解strlen函数的参数以及返回值&#xff0c;还有使用方法。 1. 计数器方法 #include <stdio.h>size_t my_strlen(const char* str) {int count 0;while (*str) {count;}return count; } int main() {char arr[] "abcdef";int len …
阅读更多...
专科论文降重修改技巧 神码ai

专科论文降重修改技巧 神码ai

大家好&#xff0c;今天来聊聊专科论文降重修改技巧&#xff0c;希望能给大家提供一点参考。 以下是针对论文重复率高的情况&#xff0c;提供一些修改建议和技巧&#xff0c;可以借助此类工具&#xff1a; 专科论文降重修改技巧 在专科论文的撰写过程中&#xff0c;查重率是一…
阅读更多...
高质量C/C++代码心得

高质量C/C++代码心得

写出高质量的C/C代码是一个涉及多方面的任务&#xff0c;它要求程序员不仅具备扎实的语言基础&#xff0c;还需要掌握一系列的软件设计和开发原则。下面将详细介绍如何写出高质量的C/C代码&#xff0c;并通过10个例子进行具体阐述。 一、编码规范 编写高质量的代码&#xff…
阅读更多...
时序预测 | Python实现LSTM-Attention-XGBoost组合模型电力需求预测

时序预测 | Python实现LSTM-Attention-XGBoost组合模型电力需求预测

时序预测 | Python实现LSTM-Attention-XGBoost组合模型电力需求预测 目录 时序预测 | Python实现LSTM-Attention-XGBoost组合模型电力需求预测预测效果基本描述程序设计参考资料预测效果 基本描述 该数据集因其每小时的用电量数据以及 TSO 对消耗和定价的相应预测而值得注意，从…
阅读更多...
力扣刷题-二叉树-二叉树左叶子之和

力扣刷题-二叉树-二叉树左叶子之和

404 左叶子之和 给定二叉树的根节点 root &#xff0c;返回所有左叶子之和。 示例 1&#xff1a; 输入: root [3,9,20,null,null,15,7] 输出: 24 解释: 在这个二叉树中&#xff0c;有两个左叶子&#xff0c;分别是 9 和 15&#xff0c;所以返回 24 思路 迭代法 迭代法理解…
阅读更多...
springboot升级到3.2导致mybatis-plus启动报错

springboot升级到3.2导致mybatis-plus启动报错

在springboot升级到3.2时&#xff0c;服务启动报错 java.lang.IllegalArgumentException: Invalid value type for attribute ‘factoryBeanObjectType’: java.lang.String&#xff1a; java.lang.IllegalArgumentException: Invalid value type for attribute factoryBeanOb…
阅读更多...
基于Java SSM框架实现水果销售网站系统项目【项目源码+论文说明】计算机毕业设计

基于Java SSM框架实现水果销售网站系统项目【项目源码+论文说明】计算机毕业设计

基于java的SSM框架实现水果销售网站系统演示 摘要 21世纪的今天&#xff0c;随着社会的不断发展与进步&#xff0c;人们对于信息科学化的认识&#xff0c;已由低层次向高层次发展&#xff0c;由原来的感性认识向理性认识提高&#xff0c;管理工作的重要性已逐渐被人们所认识&a…
阅读更多...
Linux 基本语句_16_Udp网络聊天室

Linux 基本语句_16_Udp网络聊天室

代码&#xff1a; 服务端代码&#xff1a; #include <stdio.h> #include <arpa/inet.h> #include <sys/types.h> #include <sys/socket.h> #include <netinet/in.h> #include <stdlib.h> #include <unistd.h> #include <string…
阅读更多...
信号量机制理论详解专题

信号量机制理论详解专题

一文学懂信号量机制的各种大题&#xff0c;详细操作见下文~ 1965年&#xff0c;荷兰学者Dijkstra提出的信号量&#xff08;Semaphores&#xff09;机制是一种卓有成效的进程同步工具。在长期且广泛的应用中&#xff0c;信号量机制又得到了很大的发展&#xff0c;它从整型信号量…
阅读更多...
LRU 缓存机制_题解（一道经典的数据结构算法题）

LRU 缓存机制_题解（一道经典的数据结构算法题）

LRU 缓存机制_题解&#xff08;一道经典的数据结构算法题&#xff09; 146. LRU 缓存 请你设计并实现一个满足 LRU (最近最少使用) 缓存 约束的数据结构。 实现 LRUCache 类&#xff1a; LRUCache(int capacity) 以 正整数 作为容量 capacity 初始化 LRU 缓存 int get(int k…
阅读更多...
最新文章

Copyright @ 2022~2023