自定义,也是源码的基本逻辑

''' {'ip1':[时间1 ,时间2], 'ip2':[时间1, ], } #(1)取出访问者ip # (2)判断当前ip不在访问字典里,添加进去,并且直接返回True,表示第一次访问,在字典里,继续往下走 # (3)循环判断当前ip的列表,有值,并且当前时间减去列表的最后一个时间大于60s,把这种数据pop掉,这样列表中只有60s以内的访问时间, # (4)判断,当列表小于3,说明一分钟以内访问不足三次,把当前时间插入到列表第一个位置,返回True,顺利通过 # (5)当大于等于3,说明一分钟内访问超过三次,返回False验证失败 ''' class MyThrottle(): visitor_dic = {} def __init__(self): self.history = None def allow_request(self, request, view): ip = request.META.get('REMOTE_ADDR') # 拿出ip地址,固定写法 # 不在字典中,说明是第一次访问 ctime = time.time() if ip not in self.visitor_dic: self.visitor_dic[ip] = [ctime, ] return True # 根据当前访问者ip,取出访问的时间列表 history = self.visitor_dic[ip] self.history = history while history and ctime - history[-1] > 60: #循环取,最后一个判断大于60,取倒数第二个, history.pop() if len(history) < 3: # 把当前时间放到第0个位置上 history.insert(0, ctime) return True return False def wait(self): # 剩余时间 ctime = time.time() return 60 - (ctime - self.history[-1])
使用
-内置的访问频率控制类SimpleRateThrottle -写一个类,继承SimpleRateThrottle -class MyThrottle(SimpleRateThrottle): scope='aaa' #这里的aaa必须与setting中的aaa对应 def get_cache_key(self, request, view): return self.get_ident(request) -在setting中: REST_FRAMEWORK = { 'DEFAULT_THROTTLE_RATES':{ 'aaa':'10/m' #与scope的对应处, } } #后边的参数分析:10是访问次数,后边的是这里边的任意一个 #{'s': 1, 'm': 60, 'h': 3600, 'd': 86400} ------------------------------------ -使用 #1、局部使用: 在视图类中写 throttle_classes = [MyThrottle,] #2、全局使用: 在setting中: REST_FRAMEWORK = { 'DEFAULT_THROTTLE_CLASSES':['app01.MyAuth.MyThrottle',], 'DEFAULT_THROTTLE_RATES':{ 'aaa':'10/m' } } #3、局部禁用: 在视图类中写 throttle_classes = [] --------------------------------------- #错误信息改成中文显示:(在视图类中写) def throttled(self, request, wait): class MyThrottled(exceptions.Throttled): default_detail = '请稍候' extra_detail_singular = '还剩 {wait} 秒.' extra_detail_plural = '还剩 {wait} 秒' raise MyThrottled(wait)
源码分析
有时间了整理