自定义频率类
逻辑如下：
# （1）取出访问者ip # （2）判断当前ip不在访问字典里，添加进去，并且直接返回True,表示第一次访问，在字典里，继续往下走 # （3）循环判断当前ip的列表，有值，并且当前时间减去列表的最后一个时间大于60s，把这种数据pop掉，这样列表中只有60s以内的访问时间， # （4）判断，当列表小于3，说明一分钟以内访问不足三次，把当前时间插入到列表第一个位置，返回True，顺利通过 # （5）当大于等于3，说明一分钟内访问超过三次，返回False验证失败
代码如下：
from rest_framework.throttling import BaseThrottle
import time
VISIT_RECORD = {} #保存访问记录
class VisitThrottle(BaseThrottle):
'''60s内只能访问3次'''
def __init__(self):
self.history = None #初始化访问记录
def allow_request(self,request,view):
#获取用户ip (get_ident)
remote_addr = self.get_ident(request)
ctime = time.time()
#如果当前IP不在访问记录里面，就添加到记录
if remote_addr not in VISIT_RECORD:
VISIT_RECORD[remote_addr] = [ctime,] #键值对的形式保存
return True #True表示可以访问
#获取当前ip的历史访问记录
history = VISIT_RECORD.get(remote_addr)
#初始化访问记录
self.history = history
#如果有历史访问记录，并且最早一次的访问记录离当前时间超过60s，就删除最早的那个访问记录，
#只要为True，就一直循环删除最早的一次访问记录
while history and history[-1] < ctime - 60:
history.pop()
#如果访问记录不超过三次，就把当前的访问记录插到第一个位置（pop删除最后一个）
if len(history) < 3:
history.insert(0,ctime)
return True
def wait(self):
'''还需要等多久才能访问'''
ctime = time.time()
return 60 - (ctime - self.history[-1])
settings.py中配置全局频率限制
#全局
REST_FRAMEWORK = {
# 频率
"DEFAULT_THROTTLE_CLASSES":['API.utils.throttle.VisitThrottle'],
}
继承内置SimpleRateThrottle类
throttle.py
from rest_framework.throttling import SimpleRateThrottle
class VisitThrottle(SimpleRateThrottle):
'''匿名用户60s只能访问三次（根据ip）'''
scope = 'NBA' #这里面的值，自己随便定义，settings里面根据这个值配置Rate
def get_cache_key(self, request, view):
#通过ip限制节流
return self.get_ident(request)
class UserThrottle(SimpleRateThrottle):
'''登录用户60s可以访问10次'''
scope = 'NBAUser' #这里面的值，自己随便定义，settings里面根据这个值配置Rate
def get_cache_key(self, request, view):
return request.user.username
settings.py
#全局
REST_FRAMEWORK = {
#节流
"DEFAULT_THROTTLE_CLASSES":['API.utils.throttle.UserThrottle'], #全局配置，登录用户节流限制（10/m）
"DEFAULT_THROTTLE_RATES":{
'NBA':'3/m', #没登录用户3/m，NBA就是scope定义的值
'NBAUser':'10/m', #登录用户10/m，NBAUser就是scope定义的值
}
}
view.py
class AuthView(APIView):
.
.
# 默认的节流是登录用户（10/m）,AuthView不需要登录，这里用匿名用户的节流（3/m）
throttle_classes = [VisitThrottle,]
错误信息提示
class Course(APIView):
authentication_classes = [TokenAuth, ]
permission_classes = [UserPermission, ]
throttle_classes = [MyThrottles,]
def get(self, request):
return HttpResponse('get')
def post(self, request):
return HttpResponse('post')
def throttled(self, request, wait):
from rest_framework.exceptions import Throttled
class MyThrottled(Throttled):
default_detail = '傻逼啊'
extra_detail_singular = '还有 {wait} second.'
extra_detail_plural = '出了 {wait} seconds.'
raise MyThrottled(wait)
源码分析
SimpleRateThrottle源码
