频率限制
一个网站的访问频率限制是非常重要的，访问频率限制做的好可以预防爬虫等恶意行为。
使用drf的频率限制对网站接口访问做出限制十分的便捷好用，你只需要直接进行配置即可。
内置限制
局部使用
首先我们在视图中进行配置：
from rest_framework.throttling import UserRateThrottle # 已登录的
from rest_framework.throttling import AnonRateThrottle # 未登录的
from rest_framework.generics import GenericAPIView
from rest_framework.response import Response
from django.contrib import auth
class ThrottleTestAPI(GenericAPIView):
throttle_classes = [UserRateThrottle,AnonRateThrottle]
def get(self,request):
if request.user.is_authenticated: # 用户一登陆
return Response(data="你的请求次数有十次每分钟")
return Response("你的请求次数只有三次每分钟")
def post(self,request):
user_obj = auth.authenticate(username="admin",password="admin123")
auth.login(request,user_obj)
return Response("登录成功了")
其次，针对已登录的用户和未登录的用户，可以在settings中进行配置限制次数：
REST_FRAMEWORK = {
'DEFAULT_THROTTLE_RATES': { # 限制次数 ， 未登录用户一分钟最多三次，登录用户最多一分钟十次
'anon': '3/m',# 会去配置的 UserRateThrottle 以及 AnonRateThrottle 中找到属性 scope ，scope对应的就是生效的配置
'user': '10/m'
}
}
全局使用
如果全局使用，则可以进行如下配置：
REST_FRAMEWORK = {
'DEFAULT_THROTTLE_CLASSES': (
'rest_framework.throttling.AnonRateThrottle','rest_framework.throttling.UserRateThrottle'
),'DEFAULT_THROTTLE_RATES': { # 限制次数 ， 未登录用户一分钟最多三次，登录用户最多一分钟十次
'anon': '3/m',# 会去配置的 UserRateThrottle 以及 AnonRateThrottle 中找到属性 scope ，scope对应的就是生效的配置
'user': '10/m'
}
}
如果想针对某一个视图取消全局配置，则将throttle_classes设置为空列表即可：
class ThrottleTestAPI(GenericAPIView):
throttle_classes = []
def get(self,request):
if request.user.is_authenticated:
return Response(data="你的请求次数有十次每分钟")
return Response("你的请求次数只有三次每分钟")
def post(self,user_obj)
return Response("登录成功了")
自定制限制
频率限制原理
drf中的频率控制基本原理是基于访问次数和时间的，当然我们可以通过自己定义的方法来实现。当我们请求进来，走到我们频率组件的时候，drf内部会有一个字典来记录访问者的IP
以这个访问者的IP为key，value为一个列表，value里面存放访问者每次访问的时间，如下：
{ IP1: [第三次访问时间，第二次访问时间，第一次访问时间]，}
把每次访问最新时间放入列表的最前面，记录这样一个数据结构后，通过什么方式限流呢
如果我们设置的是10秒内只能访问5次，
- 判断访问者的
IP是否在这个请求IP的字典里 - 保证这个列表里都是最近10秒内的访问的时间。判断当前请求时间和列表里最早的(也就是最后的)请求时间的如果差大于10秒，说明请求以及不是最近10秒内的，删除掉继续判断倒数第二个，直到差值小于10秒
- 判断列表的长度(即访问次数)，是否大于我们设置的5次，如果大于就限流，否则放行，并把时间放入列表的最前面。
自定义限制
使用自定义限制时，需要创建一个类并且重写allow_request()以及wait()方法。
allow_request()有两个额外的参数，分别是二次包装后的request对象，以及实例化过后的视图类本身，当频率限制通过后返回True，否则返回False。
wait()方法是在return False后触发，必须返回一个int类型的值来回复频率限制还有多久取消。
class RequestLimit:
request_dict = {}
def __init__(self):
self.expiration = None
self.count = 3 # 设定最大访问次数
self.seconds = 10 # 设定过期时间，秒为单位
def allow_request(self,request,view): # 自动调用该方法
# 拿出IP
ip = request.Meta.get("REMOTE_ADDR")
import time
current_time = time.time()
# 如果ip不在字典中，则添加即可，代表当前已访问了一次
if not ip in self.request_dict:
self.request_dict[ip] = [current_time]
return True
# 如果在，判断长度是否等于设定的最大访问次数
if len(self.request_dict[ip]) == self.count:
# 如果等于最大访问次数，则判断最后一位的时间和当前时间相差是否大于指定的过期时间
if current_time - self.request_dict[ip][-1] > self.seconds:
# 如果大于，清空写入
self.request_dict[ip].clear()
self.request_dict[ip].append(current_time)
return True
else:
# 如果不大于，说明时间还没过，暂时不能访问，设置多少秒后才能访问
self.expiration = self.request_dict[ip][-1] + self.seconds
return False
# 如果在，长度不大于3，则追加当前时间
self.request_dict[ip].append(current_time)
return True
def wait(self):
import time
current_time = time.time()
result = self.expiration - current_time # 用过期的时间，减去当前的时间
return result
局部使用
直接进行局部使用即可，不要再到项目全局文件夹下的settings.py中做额外的配置了。
class ThrottleTestAPI(GenericAPIView):
throttle_classes = [app01.app01_throttle.RequestLimit] # 直接使用即可
def get(self,request):
return Response("get...")
全局使用
全局使用也不用再规定过期时间，直接在settings.py中配置使用即可：
REST_FRAMEWORK = {
'DEFAULT_THROTTLE_CLASSES':['app01.app01_throttle.RequestLimit',],}
源码分析
上面提到了内置频率的限制原理，其实针对一登陆用户也就是UserRateThrottle来说，它是对用户id进行限制。
注释里已经说的很清楚了，如何区分是ip还是id限制，则是get_cache_key()这个方法。可以看见，UserRateThrottle是id限制。
class UserRateThrottle(SimpleRateThrottle):
"""
Limits the rate of API calls that may be made by a given user.
The user id will be used as a unique cache key if the user is
authenticated. For anonymous requests,the IP address of the request will
be used.
"""
scope = 'user'
def get_cache_key(self,view):
if request.user.is_authenticated:
ident = request.user.pk
else:
ident = self.get_ident(request)
return self.cache_format % {
'scope': self.scope,'ident': ident
}
下面我们来看它是如何实现的频率限制，首先我们知道，当进行频率限制时是会对SimpleRateThrottle进行实例化，那么就SimpleRateThrottle这个类中做了什么事情。
class SimpleRateThrottle(BaseThrottle):
cache = default_cache # Django缓存
timer = time.time # 当前时间
cache_format = 'throttle_%(scope)s_%(ident)s' # 格式化后的字符串，提示频率限制还有多久
scope = None # UserRateThrottle中有，就是user
THROTTLE_RATES = api_settings.DEFAULT_THROTTLE_RATES # 默认的配置信息
def __init__(self):
if not getattr(self,'rate',None): # 找rate，显然没有
self.rate = self.get_rate() # 运行这里
self.num_requests,self.duration = self.parse_rate(self.rate)
接下来就看get_rate()：
def get_rate(self):
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)
默认配置信息：
'DEFAULT_THROTTLE_RATES': {
'user': None,'anon': None,}, 由于我们会进行全局配置，所以会去全局项目文件夹下的settings.py中找：
'DEFAULT_THROTTLE_RATES': { # 限制次数 ， 未登录用户一分钟最多三次，登录用户最多一分钟十次
'anon': '3/m',# 会去配置的 UserRateThrottle 以及 AnonRateThrottle 中找到属性 scope ，scope对应的就是生效的配置
'user': '10/m'
}
也就是说，它会返回10/m:
class SimpleRateThrottle(BaseThrottle):
cache = default_cache # Django缓存
timer = time.time # 当前时间
cache_format = 'throttle_%(scope)s_%(ident)s' # 格式化后的字符串，提示频率限制还有多久
scope = None # UserRateThrottle中有，就是user
THROTTLE_RATES = api_settings.DEFAULT_THROTTLE_RATES # 默认的配置信息
def __init__(self):
if not getattr(self,None): # 找rate，显然没有
self.rate = self.get_rate() # 运行这里 返回10/m
self.num_requests,self.duration = self.parse_rate(self.rate) 接着向下，执行。
接下来看self.parse_rate(self.rate)
def parse_rate(self,rate):
if rate is None:
return (None,None)
num,period = rate.split('/') # 进行拆分， num = 10 period=m
num_requests = int(num) # 转换数字
duration = {'s': 1,'m': 60,'h': 3600,'d': 86400}[period[0]] # 取'm'这个的第0个，就是m
return (num_requests,duration) 返回元组，(10,60)
实例化至此完成，当有请求来时，将会自动执行allow_request()这个方法，可以在APIView这里面的dispatch()中的initial()找到check_throttles()方法，它会执行allow_request()方法。
def check_throttles(self,request):
throttle_durations = []
for throttle in self.get_throttles():
if not throttle.allow_request(request,self):
throttle_durations.append(throttle.wait())
接下来我们来看SimpleRateThrottle中的allow_request()会怎么做：
def allow_request(self,view):
if self.rate is None: # 不执行
return True
self.key = self.get_cache_key(request,view) # 用户id
if self.key is None: # 不执行
return True
self.history = self.cache.get(self.key,[]) # 缓存中获取，如果获取不到这个用户id，就做一个列表。类似于{pk:[]}
self.now = self.timer() # 获取当前时间
while self.history and self.history[-1] <= self.now - self.duration: # 如果[]为真并且[]中的最后一位时间小于或等于设定的时间 60
self.history.pop() # 弹出最后一位
if len(self.history) >= self.num_requests: # 判断[]的长度是否大于等于设定的长度
return self.throttle_failure() # 验证失败
return self.throttle_success() # 验证成功
当验证成功后，它会进行在列表中插入时间：
def throttle_success(self):
self.history.insert(0,self.now) # [当前时间]
self.cache.set(self.key,self.history,self.duration) # 缓存中存放 {pk:[],超时时间}，过了时间自动清除
return True # 返回True
当验证失败后，接着回来看APIView这里面的dispatch()中的initial()的check_throttles()方法
def check_throttles(self,self): # 验证失败
throttle_durations.append(throttle.wait()) # 执行wait()
好了，源码分析看到这一步就大概差不多了。
原文链接：/django/994713.html