Django自定义视图类及实现请求参数和返回参数加解密
2023/5/25 11:22:13
本文主要是介绍Django自定义视图类及实现请求参数和返回参数加解密，对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！
django rest_framework中GenericAPIView配合拓展类mixin或者视图集viewset可以复用其代码，减少自己编写的代码量。下面我要实现自己的视图类，以减少代码量 新建一个myView.py
from collections import OrderedDict
from rest_framework import status
from rest_framework.generics import GenericAPIView
from rest_framework.pagination import PageNumberPagination
from rest_framework.response import Response
from rest_framework.settings import api_settings
from rest_framework.views import APIView
class MyView(APIView):
queryset = None # 模型数据
serializer_class = None # 序列化类
filter_class = [] # 过滤字段
lookup_field = 'id' # 删改的查找字段
ordeing_field = ('id',) # 排序字段
pagination_class = None # 分页器
def get_queryset(self):
"""
获取queryset数据列表
"""
assert self.queryset is not None, (
"'%s' should either include a `queryset` attribute, "
"or override the `get_queryset()` method."
% self.__class__.__name__
)
self.queryset = self.queryset.all()
# self.filter_queryset()
return self.queryset
def filter_queryset(self, queryset):
"""
过滤queryset数据
"""
search_dict = {}
for fc in self.filter_class:
field = self.request.query_params.dict().get(fc)
if field:
search_dict[fc] = field
queryset = queryset.filter(**search_dict)
return queryset
def get_serializer(self, *args, **kwargs):
"""
获取序列化模型
"""
serializer_class = self.get_serializer_class()
kwargs.setdefault('context', self.get_serializer_context())
return serializer_class(*args, **kwargs)
def get_serializer_class(self):
"""
获取序列化模型类， 判断存在否
"""
assert self.serializer_class is not None, (
"'%s' should either include a `serializer_class` attribute, "
"or override the `get_serializer_class()` method."
% self.__class__.__name__
)
return self.serializer_class
def get_serializer_context(self):
return {
'request': self.request,
'format': self.format_kwarg,
'view': self
}
def create(self, request, *args, **kwargs):
"""
创建一条数据
"""
serializer_class = self.get_serializer_class()
serializer = serializer_class(data=request.data)
serializer.is_valid(raise_exception=True)
serializer.save()
return Response(serializer.data, status=status.HTTP_201_CREATED)
def update(self, request, *args, **kwargs):
"""
更新修改一条数据
"""
try:
instance = self.queryset.get(id=request.data.get(self.lookup_field))
except Exception:
return Response({'state': 'fail', 'msg': '未找到该数据'}, status=status.HTTP_400_BAD_REQUEST)
serializer_class = self.get_serializer_class()
serializer = serializer_class(instance=instance, data=request.data)
serializer.is_valid(raise_exception=True)
serializer.save()
return Response(serializer.data, status=status.HTTP_201_CREATED)
def destroy(self, request, *args, **kwargs):
"""
删除数据，传入的时数组，表示可以删除多条
"""
try:
instance = self.queryset.filter(id__in=request.data.get(self.lookup_field))
except Exception:
return Response({'state': 'fail', 'msg': '未找到数据'}, status=status.HTTP_400_BAD_REQUEST)
instance.delete()
return Response({'state': 'success', 'msg': '删除成功'}, status=status.HTTP_204_NO_CONTENT)
@property
def paginator(self):
"""
分页器属性
"""
if not hasattr(self, '_paginator'):
if self.pagination_class is None:
self._paginator = None
else:
self._paginator = self.pagination_class()
return self._paginator
def get_paginate_queryset(self, queryset):
"""
获取分页queryset
"""
paginator = self.paginator
if paginator is None:
return None
return paginator.paginate_queryset(
queryset=queryset,
request=self.request,
view=self
)
def get_paginated_response(self, data):
"""
获取分页后的返回
"""
assert self.paginator is not None
# print(self.paginator.page.paginator.count)
return self.paginator.get_paginated_response(data)
def order_by_queryset(self, queryset):
"""
queryset数据进行排序
"""
return queryset.order_by(*self.ordeing_field)
class MyPagination(PageNumberPagination):
page_size = 10 # 表示每页的默认显示数量
max_page_size = 50 # max_page_size：表示每页最大显示数量，做限制使用，避免突然大量的查询数据，数据库崩溃
page_size_query_param = 'page_size' # page_size_query_param：表示url中每页数量参数
page_query_param = 'page_num' # page_query_param：表示url中的页码参数
def get_paginated_response(self, data):
"""
重构分页返回的数据
"""
return Response(OrderedDict([
('total', self.page.paginator.count),
('data', data)
]))
class MixinGetPageList:
"""
get只获取分页数据
"""
def get(self, request, *args, **kwargs):
queryset = self.get_queryset()
queryset = self.filter_queryset(queryset)
queryset = self.order_by_queryset(queryset)
serializer_class = self.get_serializer_class()
queryset = self.get_paginate_queryset(queryset)
serializer = serializer_class(queryset, many=True)
return self.get_paginated_response(serializer.data)
class MixinGetAllList:
"""
get只获取所有数据
"""
def get(self, request, *args, **kwargs):
queryset = self.get_queryset()
queryset = self.filter_queryset(queryset)
queryset = self.order_by_queryset(queryset)
serializer_class = self.get_serializer_class()
serializer = serializer_class(queryset, many=True)
return Response(serializer.data)
class MixinGetList:
"""
get获取分页和所有数据
"""
all_serializer_class = None # 获取所有的序列化类
def get(self, request, *args, **kwargs):
queryset = self.get_queryset()
queryset = self.filter_queryset(queryset)
queryset = self.order_by_queryset(queryset)
params = request.query_params.dict()
if self.pagination_class is not None and params.get('all') is None:
serializer_class = self.get_serializer_class()
queryset = self.get_paginate_queryset(queryset)
serializer = serializer_class(queryset, many=True)
return self.get_paginated_response(serializer.data)
self.serializer_class = self.all_serializer_class
serializer_class = self.get_serializer_class()
serializer = serializer_class(queryset, many=True)
return Response(serializer.data)
class MixinPostCreateModel:
"""
post增加数据
"""
def post(self, request):
return self.create(request)
class MixinPutUpdateModel:
"""
put修改数据
"""
def put(self, request):
return self.update(request)
class MixinDeleteDestroyModel:
"""
delete删除数据
"""
def delete(self, request):
return self.destroy(request)
class MyMixin(MyView):
"""
增删改查
"""
def get(self, request, *args, **kwargs):
queryset = self.get_queryset()
queryset = self.filter_queryset(queryset)
queryset = self.order_by_queryset(queryset)
serializer_class = self.get_serializer_class()
if self.pagination_class is not None:
queryset = self.get_paginate_queryset(queryset)
# print(queryset)
serializer = serializer_class(queryset, many=True)
return self.get_paginated_response(serializer.data)
serializer = serializer_class(queryset, many=True)
return Response(serializer.data)
def post(self, request):
return self.create(request)
def put(self, request):
return self.update(request)
def delete(self, request):
return self.destroy(request)
定义的数据库模型，也就是models.py的模型
class Gender(models.Model):
class Meta:
db_table = 'gender'
name = models.CharField(verbose_name="名字", max_length=16)
序列化文件 serializer.py
class GenderSerializer(serializers.ModelSerializer):
class Meta:
model = models.Gender
fields = '__all__'
def create(self, validated_data):
res = models.Gender.objects.create(**validated_data)
return res
def update(self, instance, validated_data):
instance.name = validated_data.get('name')
instance.save()
return instance
然后新建一个视图文件
from ani import models
from ani.serializer import GenderSerializer
from utils.myView import MyPagination, MyView, MixinGetList, MixinPostCreateModel, MixinPutUpdateModel, \
MixinDeleteDestroyModel
class GenderView(MyView, MixinGetList, MixinPostCreateModel, MixinPutUpdateModel, MixinDeleteDestroyModel):
queryset = models.Gender.objects.all()
serializer_class = GenderSerializer
all_serializer_class = GenderSerializer
filter_class = ['name__icontains']
pagination_class = MyPagination
lookup_field = 'id'
ordeing_field = ('-id',)
依次继承get、post、put、delete，实现查、增、改、删。
接下来对请求参数，及返回参数进行加密，加解密可以看我之前的文章
先新建一个MyResponse.py，自定义自己的返回类
import json
from rest_framework.response import Response
from utils.encryption import setDataAes
class AESResponse(Response):
def __init__(self, data=None, secret='www', status=None, template_name=None, headers=None,
exception=False, content_type=None):
enaes_data = setDataAes(secret, json.dumps(data))
super(AESResponse, self).__init__(data=enaes_data, status=status, template_name=template_name, headers=headers,
exception=exception, content_type=content_type)
将myView.py中的Response都替换为自定义返回类，新建了一个myViewEncryp.py
from collections import OrderedDict
from rest_framework import status
from rest_framework.generics import GenericAPIView
from rest_framework.pagination import PageNumberPagination
from rest_framework.settings import api_settings
from rest_framework.views import APIView
from utils.MyResponse import AESResponse
from utils.tools import get_secret
class MyView(APIView):
queryset = None # 模型数据
serializer_class = None # 序列化模型
filter_class = [] # 过滤字段
lookup_field = 'id' # 删改的查找字段
ordeing_field = ('id',) # 排序字段
pagination_class = None # 分页器
def get_queryset(self):
"""
获取queryset数据列表
"""
assert self.queryset is not None, (
"'%s' should either include a `queryset` attribute, "
"or override the `get_queryset()` method."
% self.__class__.__name__
)
self.queryset = self.queryset.all()
# self.filter_queryset()
return self.queryset
def filter_queryset(self, queryset):
"""
过滤queryset数据
"""
search_dict = {}
for fc in self.filter_class:
field = self.request.query_params.dict().get(fc)
if field:
search_dict[fc] = field
queryset = queryset.filter(**search_dict)
return queryset
def get_serializer(self, *args, **kwargs):
"""
获取序列化模型
"""
serializer_class = self.get_serializer_class()
kwargs.setdefault('context', self.get_serializer_context())
return serializer_class(*args, **kwargs)
def get_serializer_class(self):
"""
获取序列化模型类， 判断存在否
"""
assert self.serializer_class is not None, (
"'%s' should either include a `serializer_class` attribute, "
"or override the `get_serializer_class()` method."
% self.__class__.__name__
)
return self.serializer_class
def get_serializer_context(self):
return {
'request': self.request,
'format': self.format_kwarg,
'view': self
}
def create(self, request, *args, **kwargs):
"""
创建一条数据
"""
serializer_class = self.get_serializer_class()
serializer = serializer_class(data=request.data)
serializer.is_valid(raise_exception=True)
serializer.save()
secret = get_secret(request)
return AESResponse(serializer.data, secret=secret, status=status.HTTP_201_CREATED)
def update(self, request, *args, **kwargs):
"""
更新修改一条数据
"""
secret = get_secret(request)
try:
instance = self.queryset.get(id=request.data.get(self.lookup_field))
except Exception:
return AESResponse({'state': 'fail', 'msg': '未找到该数据'}, secret=secret, status=status.HTTP_400_BAD_REQUEST)
serializer_class = self.get_serializer_class()
serializer = serializer_class(instance=instance, data=request.data)
serializer.is_valid(raise_exception=True)
serializer.save()
return AESResponse(serializer.data, secret=secret, status=status.HTTP_201_CREATED)
def destroy(self, request, *args, **kwargs):
"""
删除数据，传入的时数组，表示可以删除多条
"""
secret = get_secret(request)
try:
instance = self.queryset.filter(id__in=request.data.get(self.lookup_field))
except Exception:
return AESResponse({'state': 'fail', 'msg': '未找到数据'}, secret=secret, status=status.HTTP_400_BAD_REQUEST)
instance.delete()
return AESResponse({'state': 'success', 'msg': '删除成功'}, secret=secret, status=status.HTTP_204_NO_CONTENT)
@property
def paginator(self):
"""
分页器属性
"""
if not hasattr(self, '_paginator'):
if self.pagination_class is None:
self._paginator = None
else:
self._paginator = self.pagination_class()
return self._paginator
def get_paginate_queryset(self, queryset):
"""
获取分页queryset
"""
paginator = self.paginator
if paginator is None:
return None
return paginator.paginate_queryset(
queryset=queryset,
request=self.request,
view=self
)
def get_paginated_response(self, data):
"""
获取分页后的返回
"""
assert self.paginator is not None
# print(self.paginator.page.paginator.count)
return self.paginator.get_paginated_response(data)
def order_by_queryset(self, queryset):
"""
queryset数据进行排序
"""
return queryset.order_by(*self.ordeing_field)
class MyPagination(PageNumberPagination):
page_size = 10 # 表示每页的默认显示数量
max_page_size = 50 # max_page_size：表示每页最大显示数量，做限制使用，避免突然大量的查询数据，数据库崩溃
page_size_query_param = 'page_size' # page_size_query_param：表示url中每页数量参数
page_query_param = 'page_num' # page_query_param：表示url中的页码参数
def get_paginated_response(self, data):
"""
重构分页返回的数据
"""
secret = get_secret(self.request)
return AESResponse(OrderedDict([
('total', self.page.paginator.count),
('data', data)
]), secret=secret)
class MixinGetPageList:
"""
get只获取分页数据
"""
def get(self, request, *args, **kwargs):
queryset = self.get_queryset()
queryset = self.filter_queryset(queryset)
queryset = self.order_by_queryset(queryset)
serializer_class = self.get_serializer_class()
queryset = self.get_paginate_queryset(queryset)
serializer = serializer_class(queryset, many=True)
return self.get_paginated_response(serializer.data)
class MixinGetAllList:
"""
get只获取所有数据
"""
def get(self, request, *args, **kwargs):
queryset = self.get_queryset()
queryset = self.filter_queryset(queryset)
queryset = self.order_by_queryset(queryset)
serializer_class = self.get_serializer_class()
serializer = serializer_class(queryset, many=True)
secret = get_secret(request)
return AESResponse(serializer.data, secret=secret)
class MixinGetList:
"""
get获取分页和所有数据
"""
all_serializer_class = None
def get(self, request, *args, **kwargs):
queryset = self.get_queryset()
queryset = self.filter_queryset(queryset)
queryset = self.order_by_queryset(queryset)
params = request.query_params.dict()
if self.pagination_class is not None and params.get('all') is None:
serializer_class = self.get_serializer_class()
queryset = self.get_paginate_queryset(queryset)
serializer = serializer_class(queryset, many=True)
return self.get_paginated_response(serializer.data)
self.serializer_class = self.all_serializer_class
serializer_class = self.get_serializer_class()
serializer = serializer_class(queryset, many=True)
secret = get_secret(request)
return AESResponse(serializer.data, secret=secret)
class MixinPostCreateModel:
"""
post增加数据
"""
def post(self, request):
return self.create(request)
class MixinPutUpdateModel:
"""
put修改数据
"""
def put(self, request):
return self.update(request)
class MixinDeleteDestroyModel:
"""
delete删除数据
"""
def delete(self, request):
return self.destroy(request)
class MyMixin(MyView):
def get(self, request, *args, **kwargs):
queryset = self.get_queryset()
queryset = self.filter_queryset(queryset)
queryset = self.order_by_queryset(queryset)
serializer_class = self.get_serializer_class()
if self.pagination_class is not None:
queryset = self.get_paginate_queryset(queryset)
# print(queryset)
serializer = serializer_class(queryset, many=True)
return self.get_paginated_response(serializer.data)
serializer = serializer_class(queryset, many=True)
secret = get_secret(request)
return AESResponse(serializer.data, secret=secret)
def post(self, request):
return self.create(request)
def put(self, request):
return self.update(request)
def delete(self, request):
return self.destroy(request)
其中加密的密钥是用户的token，或者写死的字符串，tools.py
def get_secret(request):
"""
获取加密的key
"""
return request.META.get('HTTP_AUTHORIZATION') or 'wchime'
如果前端的请求参数加密，那么需要对参数解密，使用装饰器，decorators.py
def request_decrypt(func):
"""
解密请求参数
只对data解密，data传入的必须是字典，不然没有update属性
"""
def wrap(request, *args, **kwargs):
data = request.data
# print(data)
secret = get_secret(request)
decrypt_data = getDataAes(secret, data.get('text'))
if decrypt_data:
data = json.loads(decrypt_data)
del request.data['text']
request.data.update(data)
# print(decrypt_data)
return func(request, *args, **kwargs)
return wrap
这时候视图文件需要装饰器解密
from django.utils.decorators import method_decorator
from ani import models
from ani.serializer import GenderSerializer
from utils.decorators import request_decrypt
from utils.myViewEncryp import MyPagination, MyView, MixinGetList, MixinPostCreateModel, MixinPutUpdateModel, \
MixinDeleteDestroyModel
@method_decorator(request_decrypt, name='put')
@method_decorator(request_decrypt, name='delete')
@method_decorator(request_decrypt, name='post')
class GenderView(MyView, MixinGetList, MixinPostCreateModel, MixinPutUpdateModel, MixinDeleteDestroyModel):
queryset = models.Gender.objects.all()
serializer_class = GenderSerializer
all_serializer_class = GenderSerializer
filter_class = ['name__icontains']
pagination_class = MyPagination
lookup_field = 'id'
ordeing_field = ('-id',)
项目文件结构
请求提交参数脚本
import json
import requests
from encryption import setDataAes, getDataAes
d = {'name': 'aaa'}
body = setDataAes("wchime", json.dumps(d))
url = 'http://127.0.0.1:4000/ani/gender'
print(body)
data = {'text': body}
res = requests.post(url, json=data)
print(res.text)
print(getDataAes("wchime", res.text))
打印结果
这篇关于Django自定义视图类及实现请求参数和返回参数加解密的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！
- 2024-12-20go-zero 框架的 RPC 服务 启动start和停止 底层是怎么实现的？-icode9专业技术文章分享
- 2024-12-19Go-Zero 框架的 RPC 服务启动和停止的基本机制和过程是怎么实现的？-icode9专业技术文章分享
- 2024-12-18怎么在golang中使用gRPC测试mock数据？-icode9专业技术文章分享
- 2024-12-15掌握PageRank算法核心！你离Google优化高手只差一步！
- 2024-12-15GORM 中的标签 gorm:"index"是什么？-icode9专业技术文章分享
- 2024-12-11怎么在 Go 语言中获取 Open vSwitch (OVS) 的桥接信息（Bridge）？-icode9专业技术文章分享
- 2024-12-11怎么用Go 语言的库来与 Open vSwitch 进行交互？-icode9专业技术文章分享
- 2024-12-11怎么在 go-zero 项目中发送阿里云短信？-icode9专业技术文章分享
- 2024-12-11怎么使用阿里云 Go SDK (alibaba-cloud-sdk-go) 发送短信？-icode9专业技术文章分享
- 2024-12-10搭建个人博客网站之一、使用hugo创建个人博客网站
