zdppy_api如何实现带参数的中间件

参考代码

import timefrom api.middleware import Middleware
from api.middleware.base import BaseHTTPMiddleware
from api.exceptions import AuthException
import jwthas_logger = False
try:from log import loggerhas_logger = True
except:passdef roleapi(has_auth_func, jwtkey=None):"""判断用户角色是否有某个接口的权限:param has_auth_func 考虑到可能使用异步的MySQL库，此函数必须是一个异步函数接受Token中解析出来的role和request中的method和path作为参数，用于判断用户对该路径是否有访问权限"""return Middleware(RoleAPIAuthMiddleware, has_auth_func=has_auth_func, jwtkey=jwtkey)class RoleAPIAuthMiddleware(BaseHTTPMiddleware):def __init__(self, has_auth_func, jwtkey, app):super().__init__(app)self.has_auth_func = has_auth_funcself.jwtkey = jwtkeyasync def dispatch(self, request, call_next):# 判断是否需要跳过校验if "/login" in str(request.url) or "/register" in str(request.url):if has_logger:logger.debug("识别到正在访问不需要校验的路径", method=request.method, url=request.url)return await call_next(request)if has_logger:msg = f"开始进行基于角色和接口的高级别权限校验：method={request.method} url={request.url}"logger.debug(msg)# 获取tokentoken = request.headers.get("Authorization")if token is None:if has_logger:logger.error("权限不足：没有携带Token", token=token)raise AuthException("权限不足：没有携带Token")# 解析tokendata = Nonetry:if isinstance(self.jwtkey, str):data = jwt.parse_token(token, key=self.jwtkey)else:data = jwt.parse_token(token)except Exception as e:if has_logger:logger.error("获取用户Token失败", headers=request.headers, token=token, error=e)raise AuthException("无效的Token")# token的结果必须是字典类型if not isinstance(data, dict):if has_logger:logger.error("Token的解析结果不是字典类型", data=data)raise AuthException("无效的Token")# 必须要有过期时间expired = data.get("expired")if expired is None:if has_logger:logger.error("该Token没有设置过期时间", data=data, expired=expired)raise AuthException("无效的Token")# 校验过期时间的类型if not (isinstance(expired, int) or isinstance(expired, float)):if has_logger:logger.error("该Token的过期时间不是数字类型", expired=expired)raise AuthException("无效的Token")# 校验是否过期now = time.time()if now > expired:if has_logger:logger.warning("Token已过期", expired=expired)raise AuthException("Token已过期")# 必须要有用户名和用户IDuserid = data.get("id")username = data.get("username")userrole = data.get("role")if not (userid or username or userrole):if has_logger:logger.error("Token中应该包含用户ID，用户名和用户角色",userid=userid,username=username,userrole=userrole,token=token,data=data,)raise AuthException("无效的Token")# 获取请求方法和请求路径base_url = request.base_urlurl = request.urlmethod = request.methodpath = str(url).replace(str(base_url), "")if has_logger:logger.debug("开始查询用户是否具备接口级别的权限",method=method,path=path,userid=userid,username=username,userrole=userrole,)has_auth = Falsetry:has_auth = await self.has_auth_func(userrole, method, path)except Exception as e:if has_logger:logger.error("查询用户接口级别权限失败",method=method,path=path,userid=userid,username=username,userrole=userrole,error=e,)raise AuthException("无效的Token")if has_logger:logger.debug("成功查询用户是否具备接口级别的权限",has_auth=has_auth,method=method,path=path,userid=userid,username=username,userrole=userrole,)if not has_auth:if has_logger:logger.warning("权限不足",has_auth=has_auth,method=method,path=path,userid=userid,username=username,userrole=userrole,)raise AuthException("权限不足")# 权限充足，发送请求，获取响应response = await call_next(request)return response

调用栈分析

使用中间件

app1 = api.Api(routes=[api.resp.get("/", index),api.resp.get("/1", index),api.resp.post("/login", login),],middleware=[# 默认是：zhangdapeng zhangdapng520# 可以传入账号和密码进行覆盖apimidauth.roleapi(has_auth_func)]
)

apimidauth.roleapi(has_auth_func) 实例化了一个中间件对象。

has_auth_func 是一个函数

这个函数，会在中间件里面，被调用到，完整代码如下。

async def has_auth_func(role, method, path):"""校验角色对method和path是否有访问权限"""print(role, method, path)if not str(path).startswith("/"):path = f"/{path}"# GET:/1auth = str(method).upper() + ":" + path# 判断是否有权限role_auth_dict = auth_dict.get(role)if not isinstance(role_auth_dict, dict):return Falseif not role_auth_dict.get(auth):return Falsereturn True

roleapi 方法

完整代码如下：

def roleapi(has_auth_func, jwtkey=None):"""判断用户角色是否有某个接口的权限:param has_auth_func 考虑到可能使用异步的MySQL库，此函数必须是一个异步函数接受Token中解析出来的role和request中的method和path作为参数，用于判断用户对该路径是否有访问权限"""return Middleware(RoleAPIAuthMiddleware, has_auth_func=has_auth_func, jwtkey=jwtkey)

这个其实就是一个普通的函数。
不过比较特殊的地方在于，这个函数的返回值是一个Middleware类的实例。这个类接收如下参数：

  • RoleAPIAuthMiddleware：自定义的中间件
  • has_auth_func：实际上是自定义中间件需要的一个参数
  • jwtkey：实际上也是自定义中间件需要的一个参数

注意：has_auth_func 和 jwtkey 这两个参数，不是 Middleware 类必须的，而是我们自定义的 RoleAPIAuthMiddleware 需要的。

RoleAPIAuthMiddleware 的代码分析

完整代码：

class RoleAPIAuthMiddleware(BaseHTTPMiddleware):def __init__(self, has_auth_func, jwtkey, app):super().__init__(app)self.has_auth_func = has_auth_funcself.jwtkey = jwtkeyasync def dispatch(self, request, call_next):# 判断是否需要跳过校验if "/login" in str(request.url) or "/register" in str(request.url):if has_logger:logger.debug("识别到正在访问不需要校验的路径", method=request.method, url=request.url)return await call_next(request)if has_logger:msg = f"开始进行基于角色和接口的高级别权限校验：method={request.method} url={request.url}"logger.debug(msg)# 获取tokentoken = request.headers.get("Authorization")if token is None:if has_logger:logger.error("权限不足：没有携带Token", token=token)raise AuthException("权限不足：没有携带Token")# 解析tokendata = Nonetry:if isinstance(self.jwtkey, str):data = jwt.parse_token(token, key=self.jwtkey)else:data = jwt.parse_token(token)except Exception as e:if has_logger:logger.error("获取用户Token失败", headers=request.headers, token=token, error=e)raise AuthException("无效的Token")# token的结果必须是字典类型if not isinstance(data, dict):if has_logger:logger.error("Token的解析结果不是字典类型", data=data)raise AuthException("无效的Token")# 必须要有过期时间expired = data.get("expired")if expired is None:if has_logger:logger.error("该Token没有设置过期时间", data=data, expired=expired)raise AuthException("无效的Token")# 校验过期时间的类型if not (isinstance(expired, int) or isinstance(expired, float)):if has_logger:logger.error("该Token的过期时间不是数字类型", expired=expired)raise AuthException("无效的Token")# 校验是否过期now = time.time()if now > expired:if has_logger:logger.warning("Token已过期", expired=expired)raise AuthException("Token已过期")# 必须要有用户名和用户IDuserid = data.get("id")username = data.get("username")userrole = data.get("role")if not (userid or username or userrole):if has_logger:logger.error("Token中应该包含用户ID，用户名和用户角色",userid=userid,username=username,userrole=userrole,token=token,data=data,)raise AuthException("无效的Token")# 获取请求方法和请求路径base_url = request.base_urlurl = request.urlmethod = request.methodpath = str(url).replace(str(base_url), "")if has_logger:logger.debug("开始查询用户是否具备接口级别的权限",method=method,path=path,userid=userid,username=username,userrole=userrole,)has_auth = Falsetry:has_auth = await self.has_auth_func(userrole, method, path)except Exception as e:if has_logger:logger.error("查询用户接口级别权限失败",method=method,path=path,userid=userid,username=username,userrole=userrole,error=e,)raise AuthException("无效的Token")if has_logger:logger.debug("成功查询用户是否具备接口级别的权限",has_auth=has_auth,method=method,path=path,userid=userid,username=username,userrole=userrole,)if not has_auth:if has_logger:logger.warning("权限不足",has_auth=has_auth,method=method,path=path,userid=userid,username=username,userrole=userrole,)raise AuthException("权限不足")# 权限充足，发送请求，获取响应response = await call_next(request)return response

基本结构分析

class RoleAPIAuthMiddleware(BaseHTTPMiddleware):def __init__(self, has_auth_func, jwtkey, app):super().__init__(app)self.has_auth_func = has_auth_funcself.jwtkey = jwtkey

自定义的中间件类，需要继承：BaseHTTPMiddleware，这个类来自于 from api.middleware.base import BaseHTTPMiddleware 。

初始化方法：

def __init__(self, has_auth_func, jwtkey, app):super().__init__(app)self.has_auth_func = has_auth_funcself.jwtkey = jwtkey

在这个初始化方法中，我们定义了此中间件需要的参数。

我们在这里定义的是类的参数，但是实际上最后传递参数的方式是：

return Middleware(RoleAPIAuthMiddleware, has_auth_func=has_auth_func, jwtkey=jwtkey)

这里的 app 是没有传参的。

核心是 dispatch 方法

async def dispatch(self, request, call_next):# 判断是否需要跳过校验if "/login" in str(request.url) or "/register" in str(request.url):if has_logger:logger.debug("识别到正在访问不需要校验的路径", method=request.method, url=request.url)return await call_next(request)if has_logger:msg = f"开始进行基于角色和接口的高级别权限校验：method={request.method} url={request.url}"logger.debug(msg)# 获取tokentoken = request.headers.get("Authorization")if token is None:if has_logger:logger.error("权限不足：没有携带Token", token=token)raise AuthException("权限不足：没有携带Token")# 解析tokendata = Nonetry:if isinstance(self.jwtkey, str):data = jwt.parse_token(token, key=self.jwtkey)else:data = jwt.parse_token(token)except Exception as e:if has_logger:logger.error("获取用户Token失败", headers=request.headers, token=token, error=e)raise AuthException("无效的Token")# token的结果必须是字典类型if not isinstance(data, dict):if has_logger:logger.error("Token的解析结果不是字典类型", data=data)raise AuthException("无效的Token")# 必须要有过期时间expired = data.get("expired")if expired is None:if has_logger:logger.error("该Token没有设置过期时间", data=data, expired=expired)raise AuthException("无效的Token")# 校验过期时间的类型if not (isinstance(expired, int) or isinstance(expired, float)):if has_logger:logger.error("该Token的过期时间不是数字类型", expired=expired)raise AuthException("无效的Token")# 校验是否过期now = time.time()if now > expired:if has_logger:logger.warning("Token已过期", expired=expired)raise AuthException("Token已过期")# 必须要有用户名和用户IDuserid = data.get("id")username = data.get("username")userrole = data.get("role")if not (userid or username or userrole):if has_logger:logger.error("Token中应该包含用户ID，用户名和用户角色",userid=userid,username=username,userrole=userrole,token=token,data=data,)raise AuthException("无效的Token")# 获取请求方法和请求路径base_url = request.base_urlurl = request.urlmethod = request.methodpath = str(url).replace(str(base_url), "")if has_logger:logger.debug("开始查询用户是否具备接口级别的权限",method=method,path=path,userid=userid,username=username,userrole=userrole,)has_auth = Falsetry:has_auth = await self.has_auth_func(userrole, method, path)except Exception as e:if has_logger:logger.error("查询用户接口级别权限失败",method=method,path=path,userid=userid,username=username,userrole=userrole,error=e,)raise AuthException("无效的Token")if has_logger:logger.debug("成功查询用户是否具备接口级别的权限",has_auth=has_auth,method=method,path=path,userid=userid,username=username,userrole=userrole,)if not has_auth:if has_logger:logger.warning("权限不足",has_auth=has_auth,method=method,path=path,userid=userid,username=username,userrole=userrole,)raise AuthException("权限不足")# 权限充足，发送请求，获取响应response = await call_next(request)return response

中间件核心方法分析

参数是什么

async def dispatch(self, request, call_next):

首先，这个方法是一个异步方法。
第一个参数是请求对象，存储了客户端的所有请求信息。
第二个参数是调用下一个中间件的对象，如果成功了，则调用此方法得到一个response对象，返回response对象即可。

如果成功了返回什么？

如果成功了，则调用此 call_next 方法得到一个response对象，返回response对象即可。

response = await call_next(request)
return response

如果失败了，该返回什么？

示例代码如下：

if has_logger:logger.error("Token的解析结果不是字典类型", data=data)
raise AuthException("无效的Token")

首先，我们是记录错误日志。
然后，抛出一个异常。
因为 zdppy_api 已经内部封装了全局异常处理，所以这个异常，最终会被全局异常错误处理器自动捕获，并返回给客户端一个比较通用且友好的错误信息。

支持哪些异常类

首先是 AuthException，这个来源于：

from api.exceptions import AuthException

通过查看源码，我们可以知道， zdppy_api 框架，目前内置了如下异常处理器：

default_exception_handlers = {404: not_found,500: server_error,HTTPException: handle_http_exception,AuthException: handle_auth_exception,Exception: handle_exception,
}

最终总结

如果，我们要实现 db 请求上下文中间件：

  • 请求开始时，自动建立连接
  • 请求结束是，自动断开连接

那么，我们的实现思路如下：

  • 1、实现一个 OrmRequestMiddleware 中间件类。这个类继承 api.middleware.base.BaseHTTPMiddleware，接收一个 db 作为参数。
  • 2、实现一个 apimidorm.request(db) 方法，这个方法的返回值是 Middleware(OrmRequestMiddleware, db=db)
  • 3、在 OrmRequestMiddleware 自定义中间件类中，重写 async def dispatch(self, request, call_next) 方法
  • 4、方法体中实现具体的逻辑。请求开始之前，调用 db.connect()，调用 response = call_next()，之后调用 db.colse()，最后返回 response。

以上是一个具体的实现思路，仅供参考。

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

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

相关文章

深度学习-07-反向传播的自动化

深度学习-07-反向传播的自动化

深度学习-07-反向传播的自动化 本文是《深度学习入门2-自製框架》 的学习笔记，记录自己学习心得，以及对重点知识的理解。如果内容对你有帮助，请支持正版，去购买正版书籍，支持正版书籍不仅是尊重作者的辛勤劳动&#xf…
阅读更多...
OJ题目【栈和队列】

OJ题目【栈和队列】

题目导入 栈： 题目一：有效的括号题目二：用栈实现队列 队列 题目：实现循环队列 栈 题目一 有效的括号 题目要求 给定一个只包括 ‘(’，‘)’，‘{’，‘}’，‘[’，‘…
阅读更多...
LeetCode746使用最小花费爬楼梯

LeetCode746使用最小花费爬楼梯

题目描述 给你一个整数数组 cost ，其中 cost[i] 是从楼梯第 i 个台阶向上爬需要支付的费用。一旦你支付此费用，即可选择向上爬一个或者两个台阶。你可以选择从下标为 0 或下标为 1 的台阶开始爬楼梯。请你计算并返回达到楼梯顶部的最低花费。 解析 动态…
阅读更多...
负载均衡加权轮询算法

负载均衡加权轮询算法

随机数加权轮询算法 public int select() {int[] weights {10, 20, 50};int totalWeight weights[0] weights[1] weights[2];// 取随机数int offset ThreadLocalRandom.current().nextInt(totalWeight);for (int i 0; i < weights.length; i) {offset - weights[i];i…
阅读更多...
七校联赛题铅笔姿态及笔迹检测装置--mpu6050识别数字

七校联赛题铅笔姿态及笔迹检测装置--mpu6050识别数字

前言 前几天打完比赛&#xff0c;收获还是挺大的&#xff0c;数字识别部分基本上浪费了绝大部分时间。先将思路简单说明。 1、题目 2、思路 针对笔迹检测&#xff0c;我们首先考虑的肯定是陀螺仪来测量加速度方向来判断书写的方向&#xff0c;从而得到书写的数字。 我们的方案…
阅读更多...
LLM的基础模型3：Transformer变种

LLM的基础模型3：Transformer变种

大模型技术论文不断&#xff0c;每个月总会新增上千篇。本专栏精选论文重点解读&#xff0c;主题还是围绕着行业实践和工程量产。若在某个环节出现卡点&#xff0c;可以回到大模型必备腔调或者LLM背后的基础模型新阅读。而最新科技&#xff08;Mamba,xLSTM,KAN&#xff09;则提…
阅读更多...
【软件测试】自动化测试如何管理测试数据

【软件测试】自动化测试如何管理测试数据

前言 在之前的自动化测试框架相关文章中&#xff0c;无论是接口自动化还是UI自动化&#xff0c;都谈及data模块和config模块&#xff0c;也就是测试数据和配置文件。 随着自动化用例的不断增加&#xff0c;需要维护的测试数据也会越来越多&#xff0c;维护成本越来越高&#…
阅读更多...
编译 freetype、sdl、sdl_ttf

编译 freetype、sdl、sdl_ttf

/* * three lib */ /* 字符串生成位图 */ http://www.libsdl.org/projects/SDL_ttf/release/ http://www.libsdl.org/download-1.2.php https://freetype.org/ /* Freetype 2.13.0 */ ./configure CCaarch64-v01c01-linux-gnu-gcc --buildaarch64 …
阅读更多...
一个简约而不简单的记账 App(一刻记账)

一个简约而不简单的记账 App(一刻记账)

前言 在两年多前, 我曾经写过一个本地化的记账 App, 当时没有想过以后的发展. 全程是本地化的, 当时主要是为了练习 Compose 而写的. TallyApp 而现在一个完整的记账 App 它来了 一刻记账 算是圆了我两年前的梦了吧. 也让我可以真正的使用自己的记账软件. 下面是 一刻记账 记…
阅读更多...
C++实现简单的线程池

C++实现简单的线程池

最近在学习音视频的时候&#xff0c;解封装和解码的时候用到了多线程。于是把线程池的知识补了一下。 线程池的这个知识点有会涉及到锁&#xff0c;生产者消费者设计模式&#xff0c;纯虚函数继承等知识。在学习的时候可以根据知识点扩展延伸。 楼主在学习线程池这部分的时候没…
阅读更多...
10 -力扣高频 SQL 50 题（基础版)

10 -力扣高频 SQL 50 题（基础版)

10 - 每台机器的进程平均运行时间 -- sum(if(activity_type end,timestamp ,-timestamp )) -- 如果activity_type为“end”,值为timestamp&#xff0c;为“start” 为-timestamp&#xff0c;所有数相加end-start -- count(distinct process_id),获取同一机器有几个进行id -- r…
阅读更多...
鸿蒙HarmonyOS实战—如何使用Video组件播放视频

鸿蒙HarmonyOS实战—如何使用Video组件播放视频

1.视频播放 鸿蒙系统中&#xff0c;关于视频播放&#xff0c;可以使用上层视频组件Video。 参数如下 src 支持file:///data/storage路径前缀的字符串&#xff0c;用于读取应用沙箱路径内的资源。需要保证目录包路径下的文件有可读权限。 说明&#xff1a;视频支持的格式是&am…
阅读更多...
为何限定项目的 Node.js 版本

为何限定项目的 Node.js 版本

首先区分三个概念nvm&#xff0c;npm&#xff0c;nodejs。 Node.js: Node.js 是一个基于 Chrome V8 引擎的 JavaScript 运行时环境。它允许开发者使用 JavaScript 在服务器端编写应用程序,而不仅限于在浏览器中运行 JavaScript。Node.js 提供了一系列内置的模块和 API,使得开发…
阅读更多...
Python中数字比较与获取较大值的深入解析

Python中数字比较与获取较大值的深入解析

目录 一、引言 二、Python数字类型概述 三、数字比较操作符 四、获取较大值的逻辑与实现 五、高级话题&#xff1a;使用内置函数和库 六、性能分析与优化 七、案例分析 八、总结与展望 一、引言 在编程世界中&#xff0c;数字的比较和获取较大值是基础且常见的操作。P…
阅读更多...
Java 获取和修改期日与时间的各种操作方法

Java 获取和修改期日与时间的各种操作方法

LocalDateTime获取当地日期和时间 import java.time.LocalDateTime; /*LocalDateTime.now() 获取当前时间*/ public class LocalDateTimeDemo {public static void main(String[] args) {LocalDateTime time1 LocalDateTime.now();System.out.println(time1);//2024-06-01T13…
阅读更多...
【python】flask相关包依赖关系问题

【python】flask相关包依赖关系问题

【背景】 做flask项目时&#xff0c;由于涉及多个包&#xff0c;比如flask&#xff0c;flask-wtf&#xff0c;werkqeug等&#xff0c;不同版本情况下&#xff0c;互相依赖关系的确实会导致报错。比如报某个依赖无法从flask的相关包中导入等。 【解决办法】 实测可用版本搭配…
阅读更多...
四川景源畅信：抖音做直播有哪些人气品类？

四川景源畅信：抖音做直播有哪些人气品类？

随着互联网科技的飞速发展&#xff0c;抖音作为新兴的社交媒体平台&#xff0c;已经成为了人们日常生活中不可或缺的一部分。而在抖音平台上&#xff0c;直播功能更是吸引了大量的用户和观众。那么&#xff0c;在抖音上做直播有哪些人气品类呢?接下来&#xff0c;就让我们一起…
阅读更多...
目标检测数据集 - 智能零售柜商品检测数据集下载「包含VOC、COCO、YOLO三种格式」

目标检测数据集 - 智能零售柜商品检测数据集下载「包含VOC、COCO、YOLO三种格式」

数据集介绍&#xff1a;智能零售柜商品检测数据集&#xff0c;真实智能零售柜监控场景采集高质量商品图片数据&#xff0c;数据集含常见智能零售柜商品图片&#xff0c;包括罐装饮料类、袋装零食类等等。数据标注标签包含 113 个商品类别&#xff1b;适用实际项目应用&#xff…
阅读更多...
C++的爬山算法

C++的爬山算法

爬山算法&#xff08;Hill Climbing Algorithm&#xff09;是一种局部搜索算法&#xff0c;它通过迭代搜索的方式寻找问题的局部最优解。在爬山过程中&#xff0c;算法总是选择当前状态邻域中最好&#xff08;即函数值最大或最小&#xff09;的状态作为下一个状态&#xff0c;直…
阅读更多...
直播美颜插件详解：美颜SDK的技术原理

直播美颜插件详解：美颜SDK的技术原理

美颜SDK作为实现美颜功能的核心技术&#xff0c;已被广泛应用于各种直播和短视频应用中。那么&#xff0c;美颜SDK究竟是如何工作的&#xff1f;它背后的技术原理又是什么&#xff1f;本文将对直播美颜插件及其技术原理进行详解。 一、美颜SDK的概述 美颜SDK包含了各种图像处…
阅读更多...
最新文章

Copyright @ 2022~2023