drf中仿照SimpleRateThrottle自定制频率类

2023-10-06 17:57:22

刚开始弄的时候不知道出现了什么问题，计时出现负数，然后无法继续访问，后来也没调整什么就可以使用了

先放总代码

from django.core.exceptions import ImproperlyConfigured

from rest_framework.throttling import SimpleRateThrottle

from rest_framework.throttling import BaseThrottle
import time

from drf_book import settings


class IPThrottle(BaseThrottle):
    # 定义成类属性所有对象都用这一个
    VISIT_DIC = {}  # 存储来访问的IP， 个人测试使用 若是用在服务器上应该存到缓存里面，而不是存在服务器上
    scope = 'ming'  # 用来查询相应频率
    THROTTLE_RATES = settings.REST_FRAMEWORK['DEFAULT_THROTTLE_RATES']

    def __init__(self):
        self.history_list = []
        self.rate = self.get_rate()
        self.num_request, self.duration = self.parse_rate(self.rate)

    def allow_request(self, request, view):
        """
        1首先取出Ip地址
        2、判断当前Ip不在访问字典中，添加进去，并且直接返回True， 表示第一次访问
        3、循环判断当前的ip的列表， 有值， 并且 当前时间减去列表的最后一个时间大于设定的时间， 把这种数据pop掉，这样列表中只有指定时间以内的访问时间，直到列表清空，这时候就相当于首次访问
        4、判断，当列表小于指定次数， 把当前时间插入到列表第一个位置，返回True顺利通过
        5、当大于等于指定次数， 说明一分钟之内访问超过3次， 返回False验证失败
        """
        ip = request.META.get('REMOTE_ADDR')  # 获取访问者的IP
        ctime = time.time()
        if ip not in self.VISIT_DIC:
            self.VISIT_DIC[ip] = [ctime, ]
            return True
        self.history_list = self.VISIT_DIC[ip]
        while True:
            if self.history_list and ctime - self.history_list[-1] > self.duration:  # 要注意可能存在列表被清空的情况
                self.history_list.pop()  # 把最后一个去掉
            else:
                break
     if len(self.history_list) < self.num_request:
            self.history_list.insert(0, ctime) # 插入到第一个位置
           
            return True
        else:
            return False

    def get_rate(self):
　　　　　# 用于获取配置文件中的频率
"""
        Determine the string representation of the allowed request rate.
        """
        if not getattr(self, 'scope', None):
            msg = ("You must set either `.scope` or `.rate` for '%s' throttle" %
                   self.__class__.__name__)
            raise ImproperlyConfigured(msg)

        try:
            return self.THROTTLE_RATES[self.scope]
        except KeyError:
            msg = "No default throttle rate set for '%s' scope" % self.scope
            raise ImproperlyConfigured(msg)

    def parse_rate(self, rate):
        """
        Given the request rate string, return a two tuple of:
        <allowed number of requests>, <period of time in seconds>
        """
        if rate is None:
            return (None, None)
        else:
            num, period = rate.split('/')
            num_request = int(num)
            duration = {'s': 1, 'm': 60, 'h': 3600, 'd': 86400}[period[0]]
            return (num_request, duration)

    def wait(self):
        """
        Returns the recommended next request time in seconds.
        """
        # 当前时间，减去列表中最后一个时间
        ctime = time.time()

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

注意：

首先要明白在频率限制中真正起作用的是一个属性allow_request(self, request, view)，自定制的类可以继承BaseThrottle，也可以不继承
所以要自定制，就要重写allow_request(self, request, view)，看一下BaseThrottle中的，看一下注释，意思就是允许访问返回Ture，反之False

 def allow_request(self, request, view):
        """
        Return `True` if the request should be allowed, `False` otherwise.
        """
        raise NotImplementedError('.allow_request() must be overridden')

函数wait的返回值是等待的时间

    def wait(self):
        """
        Optionally, return a recommended number of seconds to wait before
        the next request.
        """
        return None

parse_rate和get_rate都是用来获取配置文件中的频率的，配置文件中的配置如下

REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_CLASSES': ['utiils.throttling.IPThrottle', ],  # 自定制频率类的路径
    'DEFAULT_THROTTLE_RATES': {
            'ming': '3/m', # duration = {'s': 1, 'm': 60, 'h': 3600, 'd': 86400}[period[0]]

}, }

小白一个，学习笔记，还请指正

码农公寓

相关文章