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

2024-06-04 22:36
文章标签 参数 实现 api 中间件 zdppy

本文主要是介绍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。

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

这篇关于zdppy_api如何实现带参数的中间件的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1031317。

相关文章

Springboot处理跨域的实现方式(附Demo)

Springboot处理跨域的实现方式(附Demo)

《Springboot处理跨域的实现方式(附Demo)》：本文主要介绍Springboot处理跨域的实现方式(附Demo),具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不... 目录Springboot处理跨域的方式1. 基本知识2. @CrossOrigin3. 全局跨域设置4.
阅读更多...
Spring Boot 3.4.3 基于 Spring WebFlux 实现 SSE 功能(代码示例)

Spring Boot 3.4.3 基于 Spring WebFlux 实现 SSE 功能(代码示例)

《SpringBoot3.4.3基于SpringWebFlux实现SSE功能(代码示例)》SpringBoot3.4.3结合SpringWebFlux实现SSE功能,为实时数据推送提供... 目录1. SSE 简介1.1 什么是 SSE？1.2 SSE 的优点1.3 适用场景2. Spring WebFlu
阅读更多...
基于SpringBoot实现文件秒传功能

基于SpringBoot实现文件秒传功能

《基于SpringBoot实现文件秒传功能》在开发Web应用时,文件上传是一个常见需求,然而,当用户需要上传大文件或相同文件多次时,会造成带宽浪费和服务器存储冗余,此时可以使用文件秒传技术通过识别重复... 目录前言文件秒传原理代码实现1. 创建项目基础结构2. 创建上传存储代码3. 创建Result类4.
阅读更多...
SpringBoot日志配置SLF4J和Logback的方法实现

SpringBoot日志配置SLF4J和Logback的方法实现

《SpringBoot日志配置SLF4J和Logback的方法实现》日志记录是不可或缺的一部分,本文主要介绍了SpringBoot日志配置SLF4J和Logback的方法实现,文中通过示例代码介绍的非... 目录一、前言二、案例一：初识日志三、案例二：使用Lombok输出日志四、案例三：配置Logback一
阅读更多...
Python如何使用__slots__实现节省内存和性能优化

Python如何使用__slots__实现节省内存和性能优化

《Python如何使用__slots__实现节省内存和性能优化》你有想过,一个小小的__slots__能让你的Python类内存消耗直接减半吗,没错,今天咱们要聊的就是这个让人眼前一亮的技巧,感兴趣的... 目录背景：内存吃得满满的类__slots__：你的内存管理小助手举个大概的例子：看看效果如何？1.
阅读更多...
Python+PyQt5实现多屏幕协同播放功能

Python+PyQt5实现多屏幕协同播放功能

《Python+PyQt5实现多屏幕协同播放功能》在现代会议展示、数字广告、展览展示等场景中,多屏幕协同播放已成为刚需,下面我们就来看看如何利用Python和PyQt5开发一套功能强大的跨屏播控系统吧... 目录一、项目概述：突破传统播放限制二、核心技术解析2.1 多屏管理机制2.2 播放引擎设计2.3 专
阅读更多...
Python实现无痛修改第三方库源码的方法详解

Python实现无痛修改第三方库源码的方法详解

《Python实现无痛修改第三方库源码的方法详解》很多时候,我们下载的第三方库是不会有需求不满足的情况,但也有极少的情况,第三方库没有兼顾到需求,本文将介绍几个修改源码的操作,大家可以根据需求进行选择... 目录需求不符合模拟示例 1. 修改源文件2. 继承修改3. 猴子补丁4. 追踪局部变量需求不符合很
阅读更多...
idea中创建新类时自动添加注释的实现

idea中创建新类时自动添加注释的实现

《idea中创建新类时自动添加注释的实现》在每次使用idea创建一个新类时,过了一段时间发现看不懂这个类是用来干嘛的,为了解决这个问题,我们可以设置在创建一个新类时自动添加注释,帮助我们理解这个类的用... 目录前言：详细操作：步骤一：点击上方的 文件（File），点击&nbmyHIgsp;设置（Setti
阅读更多...
SpringBoot实现MD5加盐算法的示例代码

SpringBoot实现MD5加盐算法的示例代码

《SpringBoot实现MD5加盐算法的示例代码》加盐算法是一种用于增强密码安全性的技术,本文主要介绍了SpringBoot实现MD5加盐算法的示例代码,文中通过示例代码介绍的非常详细,对大家的学习... 目录一、什么是加盐算法二、如何实现加盐算法2.1 加盐算法代码实现2.2 注册页面中进行密码加盐2.
阅读更多...
MySQL大表数据的分区与分库分表的实现

MySQL大表数据的分区与分库分表的实现

《MySQL大表数据的分区与分库分表的实现》数据库的分区和分库分表是两种常用的技术方案,本文主要介绍了MySQL大表数据的分区与分库分表的实现,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有... 目录1. mysql大表数据的分区1.1 什么是分区？1.2 分区的类型1.3 分区的优点1.4 分
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2