DRF自定制频率

基于IP的访问频率控制

step 1 首先写一个自定义频率控制类

# 思路
（1）判断请求是否超过限制次数，超过返回False，没有超过则返回True;
（2）限次后，需要等待多少秒才能继续访问。

# 核心业务逻辑
 取出访问者的IP，判断：
         <1>访问中的IP不在字典VISIT_DICT中，添加进去 ，直接返回True，表示第一次访问;
         <2>有值的话，拿当前的时间减去列表的最后一个时间，若大于60秒，把最后一个值给POP掉，让列表中只有60秒的访问时间
         <3>判断当前列表小于3，说明访问的次数不足3次，把当前的时间插入到列表的第一个位置，返回True
         <4>当大于等于3，说明一分钟内访问超过三次，返回False验证失败

代码实现:

class IPThrottle()
	# 定义类属性，所有的对象都可以使用
	VISIT_DICT = {}
    def __init__(self):
        self.history_list = []
    def allow_request(self,request,view):
    	remote_addr = request.META.get(‘REMOTE_ADDR‘)
        ctime = time.time()
        if remote_addr not in self.VISIT_DICT:
            self.VISIT_DICT[remote_addr] = [ctime,]
            return True
        self.history_list = self.VISIT_DICT[remote_addr]
        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):
        rest_time = 60 -(time.time()-self.history_list[-1])
        return rest_time    

全局使用：

# 全局配置
REST_FRAMEWORK = {
    ‘DEFAULT_THROTTLE_CLASSES‘:[‘utils.throttling.IPThrottle‘,],
}

局部使用：

# 在视图中设置
class QueryUserView(GenericViewSet, RetrieveModelMixin):
    queryset = User.objects.all()
    serializer_class = UserSerializer
    pk = None
    throttle_classes = [IPThrottle, ]

SimpleRateThrottle源码分析

# from rest_framework.throttling import SimpleRateThrottle

    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]  # scope：‘user‘ => ‘3/min‘
        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)
        #3  mmmmm
        num, period = rate.split(‘/‘)  # rate：‘3/min‘
        num_requests = int(num)
        duration = {‘s‘: 1, ‘m‘: 60, ‘h‘: 3600, ‘d‘: 86400}[period[0]]
        return (num_requests, duration)
    def allow_request(self, request, view):
        if self.rate is None:
            return True
        #当前登录用户的ip地址
        self.key = self.get_cache_key(request, view)  # key：‘throttle_user_1‘
        if self.key is None:
            return True

        # 初次访问缓存为空，self.history为[]，是存放时间的列表
        self.history = self.cache.get(self.key, [])
        # 获取一下当前时间，存放到 self.now
        self.now = self.timer()

        # Drop any requests from the history which have now passed the
        # throttle duration

        # 当前访问与第一次访问时间间隔如果大于60s，第一次记录清除，不再算作一次计数
        # 10 20 30 40
        # self.history:[10:23,10:55]
        # now:10:56
        while self.history and  self.now - self.history[-1] >= self.duration:
            self.history.pop()

        # history的长度与限制次数3进行比较
        # history 长度第一次访问0，第二次访问1，第三次访问2，第四次访问3失败
        if len(self.history) >= self.num_requests:
            # 直接返回False，代表频率限制了
            return self.throttle_failure()

        # history的长度未达到限制次数3，代表可以访问
        # 将当前时间插入到history列表的开头，将history列表作为数据存到缓存中，key是throttle_user_1，过期时间60s
        return self.throttle_success()

DRF自定制频率

原文：https://www.cnblogs.com/surpass123/p/13305299.html