频率限制

• 频率限制
  • 1、内置的频率限制
    • 未登录用户访问频次
    • 登录用户访问频次
    • 根据IP进行频率限制
  • 2、自定义频率限制

频率限制

可以对接口访问的频次进行限制，以减轻服务器压力
一般用于付费购买次数，投票等场景使用

1、内置的频率限制

未登录用户访问频次

全局使用:限制未登录用户1分钟访问五次

#settings.py
REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_CLASSES': ['rest_framework.throttling.AnonRateThrottle'],
    'DEFAULT_THROTTLE_RATES': {
       
        'anon': '5/m',
    },
}

# 演示全局未登录用户访问频率
class Test4View(APIView):
    authentication_classes = []
    permission_classes = []

    def get(self, request, *args, **kwargs):
        return Response('我是未登录用户')

访问5次后就会出现下图所示:

内置使用:局部未登录用户的访问频次

from rest_framework.throttling import AnonRateThrottle


# 演示局部未登录用户的访问频次
class Test5View(APIView):
    authentication_classes = []
    permission_classes = []
    throttle_classes = [AnonRateThrottle]

    def get(self, request, *args, **kwargs):
        return Response('我是未登录用户')

内置使用要注释掉全局配置中的下面这句代码

REST_FRAMEWORK = {
   # 'DEFAULT_THROTTLE_CLASSES': ['rest_framework.throttling.AnonRateThrottle'],
   
}

登录用户访问频次

全局使用

#settings.py

REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_CLASSES': [    
    'rest_framework.throttling.UserRateThrottle'
    ],
    'DEFAULT_THROTTLE_RATES': {
        'user': '10/m',
    },
}

未登录用户一分钟访问5次，登录用户一分钟访问10次

REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_CLASSES': [
        'rest_framework.throttling.AnonRateThrottle',
        'rest_framework.throttling.UserRateThrottle'
    ],
    'DEFAULT_THROTTLE_RATES': {
        'user': '10/m',
        'anon': '5/m',
    },
}

局部配置，
	再视图类中配一个就行，禁用的话加中括号就行

# 未登录用户一分钟访问5次，登录用户一分钟访问10次
from rest_framework.authentication import SessionAuthentication

class Test6View(APIView):
    authentication_classes = [SessionAuthentication]
    # permission_classes = []
    # throttle_classes = [AnonRateThrottle]

    def get(self, request, *args, **kwargs):
        return Response('我是未登录用户')

根据IP进行频率限制

#全局配置
REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_RATES': {
        'zhao': '5/m',
    },

}

#utils.throttling.py

#写一个类，继承SimpleRateThrottle，只需要重写get_cache_key方法
from rest_framework.throttling import SimpleRateThrottle


class Mythrottling(SimpleRateThrottle):
    scope = 'zhao'

    def get_cache_key(self, request, view):
        print(request.META.get('REMOTE_ADDR'))
        return request.META.get('REMOTE_ADDR')#返回什么就根据什么来限制

# python manage.py runserver 0.0.0.0:8080 局域网可以相互访问

#settings.py
ALLOWED_HOSTS = [*]

path('books3/', views.BookView.as_view()),
#views.py
from utils.throttling import Mythrottling
class BookView(APIView):
    throttle_classes = [Mythrottling]
    def get(self, request, *args, **kwargs):
        book_list = Book.objects.all()
        # 实例化得到分页器对象
        page_cursor = MyPageNumberPagination()
        book_list = page_cursor.paginate_queryset(book_list, request, view=self)
        next_url = page_cursor.get_next_link()  # 下一页
        pre_url = page_cursor.get_previous_link()  # 上一页
        print(next_url)
        print(pre_url)
        book_ser = BookModelSerializer(book_list, many=True)
        return Response(data=book_ser.data)

2、自定义频率限制

#子定制频率类需要写两个方法
	-判断是否限次，没有限次可以请求True，限次了不可以请求False 
        def allow_request(self, request, view):
   -限次后调用，显示还需要等待多长时间才能再次访问，返回等待的时间seconds
	   def wait(self):

#utils.throttling.py中
# 自定义逻辑
# （1）取出访问者IP
# (2)判断当前IP在不在访问字典里，不再就添加进去，并且直接返回True，表示第一次访问，在字典里就继续往下走
# （3）循环判断当前IP的列表，有值，并且当前时间减去列表的最后一个时间大于60s，把这种数据pop掉，这样列表中只有60s以内的访问
# （4）判断，当列表小于3，说明一分钟以内访问不足三次，把当前时间插入到列表第一个位置，返回True，顺利通过
# （5)当大于等于3，说明一分钟内访问超过三次 ，返回False,验证失败
class IPThrottle():
    VISIT_DIC = {}  # 定义成类属性，所有的对用用的都是这一个

    def __init__(self):
        self.history_list = []

    def allow_request(self, request, view):
        ctime = time.time()
        ip = request.META.get("REMOTE_ADDR")
        if ip not in self.VISIT_DIC:
            self.VISIT_DIC[ip] = [ctime, ]
            return True
        self.history_list = self.VISIT_DIC[ip]  # 当前访问者的时间列表
        while True:
            if ctime - self.history_list[-1] > 60:
                self.history_list.pop()  # 把最后一个移除
            else:
                break
        if len(self.history_list) < 3:
            self.history_list.insert(0, ctime)
            return True
        else:
            return False

    def wait(self):
        # 当前时间减去列表中最后一个时间
        ctime = time.time()

        return 60 - (ctime - self.history_list[-1])

全局配置

REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_CLASSES':(
        'utils.throttling.IPThrottle',
    ),
}