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

news2024/11/24 3:49:41

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.coloradmin.cn/o/1310638.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

slurm 23.11.0集群 debian 11.5 安装

slurm 23.11.0集群 debian 11.5 安装

slurm 23.11.0集群 debian 11.5 安装 用途 Slurm(Simple Linux Utility for Resource Management， http://slurm.schedmd.com/ )是开源的、具有容错性和高度可扩展的Linux集群超级计算系统资源管理和作业调度系统。超级计算系统可利用Slurm对资源和作业进行管理&a…
阅读更多...
变电站蓄电池在线监测系统（论文+源码）

变电站蓄电池在线监测系统（论文+源码）

1. 系统设计 本次课题为变电站蓄电池在线监测系统的设计，其系统架构如图3.1所示，包括了主控制器STC89C52单片机，液晶显示器LCD1602,模数转换器ADC0832，电流传感器ACS712，分压电阻，蜂鸣器以及温度传感器。在…
阅读更多...
Amazon SageMaker: 拓展机器学习边界，塑造未来创新趋势

Amazon SageMaker: 拓展机器学习边界，塑造未来创新趋势

授权说明：本篇文章授权活动官方亚马逊云科技文章转发、改写权，包括不限于在 亚马逊云科技开发者社区, 知乎，自媒体平台，第三方开发者媒体等亚马逊云科技官方渠道。 近期在 re:Invent 2023 大会上，亚马逊云科技发布了一…
阅读更多...
如何将Galaxybase图数据库应用于电力设备管理

如何将Galaxybase图数据库应用于电力设备管理

导读 近日，受强冷空气影响，部分北方地区出现不同程度的降雪，并持续降温。据国家电网发布的预警通知，要求启动预警响应和应急机制，密切跟踪灾害预警信息和应急响应情况，滚动研判分析覆冰、积雪、低温等对电…
阅读更多...
vite+vue3+electron搭建项目

vite+vue3+electron搭建项目

编辑器使用vscode，打开一个空文件夹 第一步 初始化vite项目 初始化vite项目，命令 npm init vite 第二步 下载依赖 进入新建的项目，下载依赖，命令 cd vite-projec npm i第三步 使用cnpm下载 electron依赖 新建一个终端&#…
阅读更多...
雪花算法详细讲解

雪花算法详细讲解

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

HNCTF

[Week1]Interesting_http 题目提示：Give me your want! POST方式传参want参数，先随便传want1； 题目问你想要什么，肯定是flag呗，传参wantflag；提示不是admin 将数据包中的Cookie：usernotadmin修…
阅读更多...
C# 实现图片的压缩和改变大小png、jpg和gif

C# 实现图片的压缩和改变大小png、jpg和gif

环境 .net6 Magick.NET-Q16-AnyCPU 13.5 Magick.NET源码 代码 using ImageMagick;namespace ImageCompress {internal class Program{static void Main(string[] args){string inputPath "imgloading.gif"; // 输入的GIF文件路径 string outputPath "im…
阅读更多...
ChatGPT Plus重新开启订阅

ChatGPT Plus重新开启订阅

12月14日凌晨，OpenAI首席执行官Sam Altman在社交平台宣布，终于找到了更多的GPU算力，重新开启订阅ChatGPT Plus。 上个月15日，OpenAI就因为算力不足，以及用户激增等原因暂停了ChatGPT Plus订阅。 Sam表示，在…
阅读更多...
nginx的location与rewrite

nginx的location与rewrite

目录 一.location 二.rewrite rewrite跳转实现： 语法格式：rewrite [flag]; flag标记说明： 三.基于域名跳转 四.基于ip跳转 五.基于旧域名跳转到新域名后面加目录 六.基于参数匹配的跳转 可以同过全局变量来匹配： 基于目…
阅读更多...
智能分析/可视化安防监控系统EasyCVR风光互补远程视频监控方案

智能分析/可视化安防监控系统EasyCVR风光互补远程视频监控方案

一、背景需求 在一些偏远地区，也具有视频监控的需求。但是这类场景中，一般无法就近获取市电，如果要长距离拉取市电，建设的成本非常高且长距离传输有安全隐患，因此风光互补远程视频监控方案的需求也较多。利用风光电转…
阅读更多...
【数组Array】力扣-303 区域和检索 - 数组不可变

【数组Array】力扣-303 区域和检索 - 数组不可变

目录 题目描述 解题过程 labuladong题解 题目描述 给定一个整数数组 nums&#xff0c;处理以下类型的多个查询: 计算索引 left 和 right &#xff08;包含 left 和 right&#xff09;之间的 nums 元素的 和 &#xff0c;其中 left < right 实现 NumArray 类&#xff…
阅读更多...
【送书活动五期】Go语言开发规范指南

【送书活动五期】Go语言开发规范指南

今天和一个小伙伴偶尔聊了两句&#xff0c;聊到现在工作的开发语言&#xff0c;大学时接触的第一个语言应该是html&#xff0c;系统且简单的学习了前端语言&#xff0c;之后伴随着学校的课程&#xff0c;C、C#、Java都有涉及&#xff0c;然后就一直已Java为主了&#xff0c;也是…
阅读更多...
工厂如何制定OEE目标

工厂如何制定OEE目标

设备综合效率OEE&#xff08;Overall Equipment Effectiveness&#xff09;是衡量工厂生产设备利用率和效率的重要指标。制定合适的OEE目标对于工厂的生产效率和效益至关重要。制定OEE目标需要综合考虑工厂的生产能力、设备状态、生产计划和质量要求等因素。下面将介绍一般工厂…
阅读更多...
【Spring】01 Bean 介绍

【Spring】01 Bean 介绍

文章目录 1. 定义2. 特性1&#xff09;可重用性2&#xff09;可配置性3&#xff09;可管理性 3. 生命周期1&#xff09;实例化2&#xff09;属性设置3&#xff09;初始化4&#xff09;使用5&#xff09;销毁 4. 配置方式1&#xff09;XML配置2&#xff09;注解配置3&#xff09…
阅读更多...
3DMax物理画笔物体填充放置绘制画笔插件安装使用方法

3DMax物理画笔物体填充放置绘制画笔插件安装使用方法

3DMax物理画笔物体填充放置绘制画笔插件&#xff0c;允许您使用笔刷以非常自然的方式用物品快速填充场景&#xff0c;并使用刚体模拟自动放置它们。 无论你是从事建筑、游戏电影还是商业。。。等等&#xff0c;你经常需要用一些物品为你的场景添加细节。手工放置它们是乏味的&…
阅读更多...
大数据技术10：Flink从入门到精通

大数据技术10：Flink从入门到精通

导语&#xff1a;前期入门Flink时&#xff0c;可以直接编写通过idea编写Flink程序&#xff0c;然后直接运行main方法&#xff0c;无需搭建环境。我碰到许多初次接触Flink的同学&#xff0c;被各种环境搭建、提交作业、复杂概念给劝退了。前期最好的入门方式就是直接上手写代码&…
阅读更多...
【ET8框架入门】2.ET框架解析

【ET8框架入门】2.ET框架解析

菜单栏相关&#xff1a;ENABLE_DLL选项 ET->ChangeDefine->ADD_ENABLE_DLL/REMOVE_ENABLE_DLL 一般在开发阶段使用Editor时需要关闭ENABLE_DLL选项。该选项关闭时&#xff0c;修改脚本之后&#xff0c;会直接重新编译所有的代码&#xff0c;Editor在运行时会直接使用最…
阅读更多...
学习pytorch18 pytorch完整的模型训练流程

学习pytorch18 pytorch完整的模型训练流程

pytorch完整的模型训练流程 1. 流程1. 整理训练数据 使用CIFAR10数据集2. 搭建网络结构3. 构建损失函数4. 使用优化器5. 训练模型6. 测试数据 计算模型预测正确率7. 保存模型 2. 代码1. model.py2. train.py 3. 结果tensorboard结果以下图片 颜色较浅的线是真实计算的值&#x…
阅读更多...
DHCP--自动获取IP地址

DHCP--自动获取IP地址

目录 一、了解DHCP服务 1、概念 2、使用DHCP的好处 3、DHCP的分配方式 二、DHCP的租约过程 1、客户机请求IP地址 2、服务器响应 3、客户机选择IP地址 4、服务器确定租约 5、服务器租约期限到了之后续期问题 6、总结 三、部署DHCP实验 1、项目要求 2、规划设计 …
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023