基于Django的Rest Framework框架的频率组件

Posted 2022-03-15 wanglei957

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了基于Django的Rest Framework框架的频率组件相关的知识，希望对你有一定的参考价值。

0|1一、频率组件的作用

　　在我们平常浏览网站的时候会发现，一个功能你点击很多次后，系统会让你休息会在点击，这其实就是频率控制，主要作用是限制你在一定时间内提交请求的次数，减少服务器的压力。

modles.py

0|1二、自定义频率组件类

#（1）取出访问者ip
# （2）判断当前ip不在访问字典里，添加进去，并且直接返回True,表示第一次访问，在字典里，继续往下走
# （3）循环判断当前ip的列表，有值，并且当前时间减去列表的最后一个时间大于60s，把这种数据pop掉，这样列表中只有60s以内的访问时间，
# （4）判断，当列表小于3，说明一分钟以内访问不足三次，把当前时间插入到列表第一个位置，返回True，顺利通过
# （5）当大于等于3，说明一分钟内访问超过三次，返回False验证失败

自定义频率组件的逻辑

myauth.py

from rest_framework.authentication import BaseAuthentication
from rest_framework.permissions import BasePermission
from app01.models import TokenUser
from rest_framework.exceptions import AuthenticationFailed
from rest_framework.throttling import BaseThrottle

class MyAuthentication(BaseAuthentication):
    #验证过程
    def authenticate(self, request):
        token = request.GET.get(‘token‘)
        token_user = TokenUser.objects.filter(token=token).first()
        if token_user:
            return token_user.user,token
        else:
            raise AuthenticationFailed(‘你还未登录,请先登录‘)


class MyPermission(BasePermission):
    message = ‘该用户的权限无法使用该功能‘
    def has_permission(self, request, view):
        type = int(request.user.type)
        if type == 2 or type ==3:
            return True
        else:
            return False


import time
class MyThrottle():
    # （1）取出访问者ip
    # （2）判断当前ip不在访问字典里，添加进去，并且直接返回True,表示第一次访问，在字典里，继续往下走
    # （3）循环判断当前ip的列表，有值，并且当前时间减去列表的最后一个时间大于60s，把这种数据pop掉，这样列表中只有60s以内的访问时间，
    # （4）判断，当列表小于3，说明一分钟以内访问不足三次，把当前时间插入到列表第一个位置，返回True，顺利通过
    # （5）当大于等于3，说明一分钟内访问超过三次，返回False验证失败
    visit_dic =   # 访问的ip时间字典
    def __init__(self):
        self.history = None   #某个ip的时间列表
    def allow_request(self, request, view):
        ip = request.META.get(‘REMOTE_ADDR‘)  #获取请求中的ip地址
        now_time = time.time()
        print(self.visit_dic)
        if ip not in self.visit_dic:
            self.visit_dic[ip] = [now_time,]  #ip第一次来时，添加ip和时间至ip时间字典
            return True
        self.history = self.visit_dic.get(ip,None)
        while self.history and (now_time-self.history[-1]>60):
            self.history.pop()
        if len(self.history)<3:
            self.history.insert(0,now_time)
            return True
        else:
            return False
    def wait(self):
        pass

views.py

from django.shortcuts import render
from rest_framework.response import Response

# Create your views here.
from rest_framework.generics import ListCreateAPIView,RetrieveUpdateDestroyAPIView
from app01 import models
from app01.myserializer import BookSerializer
from rest_framework.views import APIView
from app01.myAuth import MyAuthentication,MyPermission,MyThrottle

class BooksView(ListCreateAPIView):
    authentication_classes = [MyAuthentication]  #加上验证的类，如果有多个，会从做到右依次验证
    permission_classes = [MyPermission]
    throttle_classes = [MyThrottle]
    queryset = models.Book.objects.all()
    serializer_class = BookSerializer


class BookView(RetrieveUpdateDestroyAPIView):
    authentication_classes = [MyAuthentication]
    queryset = models.Book.objects.all()
    serializer_class = BookSerializer


import uuid
#登录
class Login(APIView):
    def post(self,request):
        name = request.data.get(‘name‘)
        pwd = request.data.get(‘pwd‘)
        user = models.User.objects.filter(name=name,pwd=pwd).first()
        if user:
            token = uuid.uuid4()
            models.TokenUser.objects.update_or_create(user=user,defaults="token":token)
            response = "status":100,"message":"登录成功"
        else:
            response = "status": 200, "message": "登录失败"
        return Response(response)

其余与权限认证代码一致。

结果1：

结果2：

0|1三、继承SimpleRateThrottle的频率组件

myauth.py

from rest_framework.authentication import BaseAuthentication
from rest_framework.permissions import BasePermission
from app01.models import TokenUser
from rest_framework.exceptions import AuthenticationFailed
from rest_framework.throttling import BaseThrottle,SimpleRateThrottle

class MyAuthentication(BaseAuthentication):
    #验证过程
    def authenticate(self, request):
        token = request.GET.get(‘token‘)
        token_user = TokenUser.objects.filter(token=token).first()
        if token_user:
            return token_user.user,token
        else:
            raise AuthenticationFailed(‘你还未登录,请先登录‘)


class MyPermission(BasePermission):
    message = ‘该用户的权限无法使用该功能‘
    def has_permission(self, request, view):
        type = int(request.user.type)
        if type == 2 or type ==3:
            return True
        else:
            return False

class MyThrottle(SimpleRateThrottle):
    scope = ‘throttle‘  #settins中的配置
    def get_cache_key(self, request, view):
        return request.META.get(‘REMOTE_ADDR‘,None)