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 多平台发布

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

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

linux性能优化-cpu使用率

linux性能优化-cpu使用率

文章目录 1.CPU使用率2.节拍率的概念2.1.查看系统节拍率2.2.用户节拍率2.3.CPU使用率公式 3.怎么查看CPU使用率3.1.top显示系统总体CPU使用情况3.2.pidstat分析每个进程CPU使用情况 4.CPU使用率过高怎么办4.1.perf命令详解 1.CPU使用率 用什么指标来描述系统的CPU性能呢?不是…
阅读更多...
matlab中Signal Builder模块的用法总结

matlab中Signal Builder模块的用法总结

目录 前言方法一方法二参考文章 前言 今天在用matlab中Signal Builder的模块时，不知道怎么去得到想要的信号源，于是上网查了一下，并记录一下 方法一 如图所示，打开自定义 上面一行是横坐标，下面一行是纵坐标 [0,1…
阅读更多...
Restrict Content Pro WordPress – 限制会员内容 付费内容网站(包含所有扩展)

Restrict Content Pro WordPress – 限制会员内容 付费内容网站(包含所有扩展)

Restrict Content Pro WordPress限制会员内容专业插件 强大的内容限制工具和强大的 WordPress 会员网站，都在一个易于管理的插件中。 购买Restrict Content Pro 最新版本并加入超过23000 名快乐客户的俱乐部。 使用 Restrict Content Pro 插件将您的独家内容锁定…
阅读更多...
【员工工资册】————大一期末答辩近满分作业分享

【员工工资册】————大一期末答辩近满分作业分享

前言 大家好吖，欢迎来到 YY 滴项目系列 ，热烈欢迎！ 本章主要内容面向接触过C语言的老铁 主要内容含： 欢迎订阅 YY滴C专栏！更多干货持续更新！以下是传送门！ PS：以下内容是部分展示&am…
阅读更多...
虚拟机下Ubuntu上网设置

虚拟机下Ubuntu上网设置

文章目录 一、虚拟机上网的两种方式1.1 NAT模式（Network Address Translation）1.2 桥接模式（Bridge Mode）1.3 简介 二、实际配置2.1 NAT模式配置2.2 桥接模式配置 之前跟着博客配了好几个也没用，后来自己慢慢模式实践测…
阅读更多...
2023/12/17 初始化

2023/12/17 初始化

普通变量（int,float,double变量）初始化： int a0; float b(0); double c0; 数组初始化： int arr[10]{0}; 指针初始化： 空指针 int *pnullptr; 被一个同类型的变量的地址初始化（赋值） int…
阅读更多...
数据结构之＜图＞的介绍

数据结构之＜图＞的介绍

图（Graph）的概念： 在数据结构中，图是由节点（顶点）和边组成的非线性数据结构。图用于表示不同对象之间的关系，其中节点表示对象，边表示对象之间的连接或关系。 1.图的基本组成元素&a…
阅读更多...
猿人学新平台第一题-魔改加密算法(js逆向)

猿人学新平台第一题-魔改加密算法(js逆向)

分析网络请求，直接抓包数据。 依旧是ajax请求，我们现在看一下参数的情况， 发现加密的算法只有token,now显然是一个时间戳，page是一个页数的显示。这里我们直接搜索这个token，查询一下他的位置。 直接找到位置&#xf…
阅读更多...
JieLink+智能终端操作平台存在弱口令漏洞

JieLink+智能终端操作平台存在弱口令漏洞

产品简介 捷顺JeLink智能终端操作平台(JSOTC2016 fJeLink)是捷顺历经多年行业经验积累，集智能硬件技术视频分析技术、互联网技术等多种技术融合，基于B/S架构，实现核心业务处理模型(用户中心、投权中心财务中心中心值班室、 运维中心车行客户…
阅读更多...
lvs-nat部署

lvs-nat部署

LVS负载均衡群集部署——NAT模式 实验环境： 负载调度器：内网关 lvs，ens33：172.16.23.10；外网关：ens36：12.0.0.1 Web服务器1：172.16.23.11 Web服务器2：172.16.23.12 NFS…
阅读更多...
Element 介绍

Element 介绍

Element 介绍 Vue 快速入门 Vue 常见组件 表格 分页组件 其他自己去看吧 链接: 其他组件
阅读更多...
Spring之容器：IOC（2）

Spring之容器：IOC（2）

学习的最大理由是想摆脱平庸，早一天就多一份人生的精彩；迟一天就多一天平庸的困扰。各位小伙伴，如果您： 想系统/深入学习某技术知识点… 一个人摸索学习很难坚持，想组团高效学习… 想写博客但无从下手，急需…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023