本文主要是介绍django rest_framework APIView类基于dispatch方法改写实现路由请求的转发,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!
在日常运维开发工作中,我们经常会去调用第三方服务接口获取数据,例如zabbix/jumpserver等等都对外提供了详细的API接口文档,可以非常方便的通过这些接口实现资源整合和devops的系统集成。
之前笔者在drf视图中处理前端请求这类第三方服务数据时,通常会预先将调用服务各类接口的方法块封装成工具类,供视图类中的方法统一调用。
通常是以下的流程 :
1.前端ajax请求----> 2.被路由映射匹配到的drf相关视图类的方法--->3.调用工具类的对应函数进行数据的获取和处理并返回给drf视图类的方法---> 4.drf视图类的方法将数据响应给前端ajax请求---> 5.前端js代码块对获取的数据进行处理并渲染到Dom上。
一个简单的demo
场景: 前端有一个table用来展示当前zabbix服务下所有的主机群组(假设不涉及主机群组的新增/更新/删除等操作,因为可能会对后端Django数据库进行操作)
这里以zabbix官方API文档提供的查询zabbix所有主机群组的接口作为例子。按照以往的思路,通常是先定义一个获取群组信息的方法(当然,我们需要先做接口的认证,获取鉴权token)。
"""
zabbix_demo.py
"""
import requests
import jsonclass ZbxApi:def __init__(self):self.url = 'http://zbx.haha.com/zabbix/api_jsonrpc.php'self.headers = {'content-type': "application/json"}self.user = 'Admin'self.password = 'hahahahahaha'@propertydef auth_get(self):body = {"jsonrpc": "2.0","method": "user.login","params": {"user": self.user,"password": self.password},"id": 1}with requests.post(self.url,data=json.dumps(body),headers=self.headers) as res:res = eval(res.text)auth = res.get('result', '')return authdef fetch_host_group(self):body = {"jsonrpc": "2.0","method": "hostgroup.get","params": {"output": "extend",},"auth": self.auth_get,"id": 1}# 步骤一:调用zabbix主机群组信息接口with requests.post(self.url,data=json.dumps(body),headers=self.headers) as response:result = eval(response.text).get('result', '')# 步骤二:将获取的数据封装成list格式group_info = list(map(lambda group: {"groupId": group.get("groupid"),"groupName": group.get("name")}, result))return group_infoif __name__ == '__main__':test = ZbxApi()print(test.fetch_host_group())# [{'groupId': '1', 'groupName': 'Templates'}, {'groupId': '2', 'groupName': 'Linux servers'}, {'groupId': '3', 'groupName': 'Zabbix servers'}, {'groupId': '4', 'groupName': 'Discovered hosts'}, {'groupId': '5', 'groupName': 'Virtual machines'}]
然后在相关视图中定义一个接口方法,当前端ajax请求该接口时,drf视图方法中再调用ZbxApi的fetch_host_group()获取主机信息,通过Response响应给前端进行渲染
from utils.zabbix_demo import ZbxApi
from rest_framework.views import APIView
from rest_framework.response import Responseclass ZbxInfo(APIView):def get(self, request):zbxapi = ZbxApi()group_list = zbxapi.fetch_host_group()return Response({'code': 1,'msg': group_info})
思考:
其实在fetch_host_group 方法的第一个步骤中,我们就已经获取到由zabbix返回的结构化数据,针对这类场景,我们可不可以简化整个前后端数据的请求和响应过程,即直接省略流程中的3和4两个过程,在django路由请求分发阶段,直接将匹配到的request转发给zabbix服务本身,再由其直接将数据响应给前端ajax请求,后端视图不再进行响应而是由前端js代码自行进行处理?
方案:
笔者在参考了网上分析drf对请求进行预分发处理过程的相关文章后,对APIView源码进行了分析发现,当请求匹配到特定视图类后,视图类在响应前,drf主要是在视图基类APIView的dispatch方法中对请求进行预处理(其实django也是在视图基类View的dispatch方法中,只不过相较于drf少了对request的认证/鉴权/限流以及APIException异常的捕获),APIView dispatch方法内容大致如下:
def dispatch(self, request, *args, **kwargs):"""`.dispatch()` is pretty much the same as Django's regular dispatch,but with extra hooks for startup, finalize, and exception handling."""self.args = argsself.kwargs = kwargs"""步骤一: 封装原生request# 对原生的request进行加工(丰富了一下功能) # request = request._request 获取原生request# authenticators = request.authenticators 获取认证类的对象# request = Request(request,parsers=self.get_parsers(),authenticators=self.get_authenticators(),negotiator=self.get_content_negotiator(),parser_context=parser_context)"""request = self.initialize_request(request, *args, **kwargs)self.request = requestself.headers = self.default_response_headers # deprecate?try:"""步骤二: 调用perform_authentication/check_permissions/check_throttles对请求实现认证,鉴权和限流"""self.initial(request, *args, **kwargs)# Get the appropriate handler method"""步骤三: 获取django view类中支持的restful 方法:['get', 'post', 'put', 'patch', 'delete', 'head', 'options', 'trace']"""if request.method.lower() in self.http_method_names:handler = getattr(self, request.method.lower(),self.http_method_not_allowed)else:handler = self.http_method_not_allowed"""步骤四: 基于反射实现跟据请求的方法不同, 执行我们在对应视图中所定义的不同的方法(即get/put/post/delete等)并将数据进行响应"""response = handler(request, *args, **kwargs)except Exception as exc:response = self.handle_exception(exc)"""步骤五: 获取匹配到的视图对应接口方法所执行的结果,即封装在Response类里的数据"""self.response = self.finalize_response(request, response, *args, **kwargs)"""步骤六: 结果数据的响应"""return self.response
理解:
从对dispatch方法的源码解读可以发现,当请求进入dispatch方法后,在依次完成一二三步骤后,在第四步通过request的method属性映射(通过getattr)到视图类对应方法上,并通过respnose = handler(request, *args, **kwargs)实例化对应方法,获取response后返回给调用方。
所以我们直接重新定义一个handler,不通过映射的方式将请求转发到视图类对应的接口方法中,而是让调用方在请求中携带fetch_host_group方法中的请求header/body/params等信息(但token 等敏感信息不要在前端携带)直接向第三方服务发起请求,dispatch在获取到数据后再直接return 给调用方,响应数据的处理也直接由调用方自行处理,django在整个过程中只扮演请求转发的角色。
实现:
1.首先改写APIView的dispatch方法并自定义一个handler(代码中的requests_handler方法)通过request模块将请求转发给第三方服务;
2.我们都知道在drf中Mixins类是用于处理dispatch分发请求并进行响应的各类接口方法的二次封装,由于我们自定义的handler已经将请求转发,所以就用不上了,但是我这里自定义一个AllowedAPICheckMixin用于校验调用方请求中携带的url以及method是否是DispatchAPIView中allowed_apis属性所允许的,希望不要于drf自带的Mixin类弄混淆;
"""
dispatchview.py
"""from rest_framework.views import APIView
import requests
from utils.zabbix import ZbxApiclass AllowedAPICheckMixin:def check(self, method, path):"""allowed_apis = [ThirdAPI('/user/<username>', ['GET', 'PUT']),]"""for third_api in self.allowed_apis:if third_api.match(method, path):return Truereturn Falseclass DispatchAPIView(APIView, AllowedAPICheckMixin):"""将请求转发给第三方服务,并返回原结果service_base_url 必选,第三方服务基础url,入zabbix api的 '127.0.0.1/zabbix'allowed_apis 必选, 所要访问的第三方服务的接口url,如 zabbix 获取主机群组的 '/api_jsonrpc.php'extra_headers 可选,可以用来存放调用第三方的 Header 字段server_name 必须,第三方服务名称,通过该字段在这个解析的过程中获取相应服务的接口鉴权token以及相应数据分析(有一些第三方服务响应根据状态码有些则是body里的errcode字段,需有区分)"""service_base_url = Noneextra_headers = Noneservice_name = Noneallowed_apis = []def dispatch(self, request, *args, **kwargs):self.args = argsself.kwargs = kwargsif not self.service_base_url:raise ValueError('第三方服务基础url不能为空')path = request.META.get('HTTP_X_SERVICE_PATH')if not path:raise ValueError('headers字段 HTTP_X_SERVICE_PATH 不能为空')# 获取请求方法(get/put/post/patch/delete/head...)method = request.methodself.check(method, path)if not self.check(method, path):raise ValueError('put api path {} in allowed_apis'.format(path))request = self.initialize_request(request, *args, **kwargs)self.request = requestself.headers = self.default_response_headers # deprecate?try:self.initial(request, *args, **kwargs)# Get the appropriate handler methodif request.method.lower() in self.http_method_names:handler = self.requests_handlerelse:handler = self.http_method_not_allowed# 所有可用方法,使用 requests_handler 处理response = handler(request, *args, **kwargs)except Exception as exc:response = self.handle_exception(exc)self.response = self.finalize_response(request, response, *args, **kwargs)return self.responsedef requests_handler(self, request, *args, **kwargs):# 获取请求方法method = request.method.lower()path = request.META.get('HTTP_X_SERVICE_PATH')# 拼接第三方服务urlurl = '{base_url}{path}'.format(base_url=self.service_base_url, path=path)# 判断传入的第三方的 Header是否为字典类型if self.extra_headers:assert isinstance(self.extra_headers, dict), ('extra_headers 需要为字典')# 获取request的params参数params = request.query_params.dict()# 获取request的bodydata = request.data# 获取第三方接口tokendata.update(self.fetch_token())# 通过requests库将请求转发给第三方服务resp = requests.request(method,url,headers=self.extra_headers,params=params,json=data,)return Response(res)@classmethoddef fetch_token(cls):if cls.service_name == 'zabbix':zbxapi = ZbxApi()data = {"auth": zbxapi.auth_get}return data
3.定义一个ThirdAPI类去封装DispatchAPIView类中的allowed_apis(可能会有多个第三方服务的接口需要被转发)属性;
"""
dispatchview.py
"""from django.urls.resolvers import _route_to_regex
import reclass ThirdAPI:def __init__(self, path, methods):for method in methods:if method != method.upper():raise ValueError('methods {} should all be UPPERCASE'.format(methods))self._path = pathself._methods = methods@propertydef path(self):return self._path@propertydef methods(self):return self._methodsdef match(self, method_to_check, path_to_check):# print(_route_to_regex(self.path), self.path, path_to_check)# 增则匹配调用方的动态路由以及请求的方法是不是在 allowed_apis 允许的范围内pattern = _route_to_regex(self.path)[0]if re.match(pattern, path_to_check) and \method_to_check.upper() in self.methods:return Truereturn False
4.相关app应用的views调用
from utils.ThirdViews import ThirdAPI, DispatchAPIView
import os
class ZbxService(DispatchAPIView):extra_headers = {'content-type': "application/json"}service_name = 'zabbix'# zabbix服务地址service_base_url = '127.0.0.1/zabbix'allowed_apis = [ThirdAPI('/api_jsonrpc.php', ['GET', 'POST'])]
5.配置一个url
from zabbix.views import ZbxService
urlpatterns = [path('api/zabbix', ZbxService.as_view()),
]
6.利用vue的axios发起调用请求
我们将需要访问的第三方的接口url放入headers的X-Service-Path, 后端DispatchAPIView类的dispatch会通过request.META去解析
<template><div><el-button @click="DispatchTest">请求分发测试</el-button></div>
</template>
<script>
import axios from 'axios'
export default {data () {return {groupInfo: []}},methods: {DispatchTest () {axios({url: 'http://127.0.0.1:8000/api/zabbix',method: 'post',params: {'id': 1},data: {'jsonrpc': '2.0','method': 'hostgroup.get','id': 1,'params': {'output': 'extend'}},headers: {'X-Service-Path': '/api_jsonrpc.php'}}).then(res => {const groups = [...res.data.result]this.groupInfo = groups.map(i => {return {groupId: i.groupid,groupName: i.name}})console.log(this.groupInfo)}).catch(err => {console.log(err)})}}
}
</script>
7.查看测试结果
这篇关于django rest_framework APIView类基于dispatch方法改写实现路由请求的转发的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!