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))

列印結果