1 频率类
写一个类，继承SimpleRateThrottle，重写get_cache_key，返回[ip，用户id]什么，就以什么做限制，编写类属性 scope = 字符串，在配置文件中配置
'DEFAULT_THROTTLE_RATES': {
'字符串': '3/m',
}
配置在视图类，全局使用(配置在配置文件中)
2 自定义频率类
源码中找---》自定义频率类如何写---》写个类，重写allow_request，如果能访问就返回True，不能访问返回False
APIView---》dispatch---》走三大认证的时候
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())
if throttle_durations:
durations = [
duration for duration in throttle_durations
if duration is not None
]
duration = max(durations, default=None)
self.throttled(request, duration)
# 3 自定义频率类 这是我们自己写的
class MyThrottles():
VISIT_RECORD = {} # 访问者 ip 字典，格式是{ip1:[时间4，时间3，时间2，时间1]，ip2:[时间，时间]，ip3:[当前时间]}
def __init__(self):
self.history = None
def allow_request(self, request, view):
# （1）取出访问者ip
# print(request.META)
ip = request.META.get('REMOTE_ADDR')
import time
ctime = time.time()
# （2）判断当前ip不在访问字典里，添加进去，并且直接返回True,表示第一次访问
if ip not in self.VISIT_RECORD:
# {ip地址作为key:[当前时间，时间1，时间2，时间3]}
self.VISIT_RECORD[ip] = [ctime, ]
return True
self.history = self.VISIT_RECORD.get(ip)
# （3）循环判断当前ip的列表，有值，并且当前时间减去列表的最后一个时间大于60s，把这种数据pop掉，这样列表中只有60s以内的访问时间，
while self.history and ctime - self.history[-1] > 60:
self.history.pop()
# （4）判断，当列表小于3，说明一分钟以内访问不足三次，把当前时间插入到列表第一个位置，返回True，顺利通过
# （5）当大于等于3，说明一分钟内访问超过三次，返回False验证失败
if len(self.history) < 3:
self.history.insert(0, ctime)
return True
else:
return False
def wait(self):
import time
ctime = time.time()
return 60 - (ctime - self.history[-1])
SimpleRateThrottle源码分析
"""
-allow_request：必须有的，频率类，必须重写它
-get_cache_key：没有具体实现，直接抛了异常，需要子类重写
-wait：必须有的，返回一个数字，告诉前端，还有多长时间能访问
-------都是为了给allow_request辅助的-----
get_rate
parse_rate
throttle_failure
throttle_success
"""
def allow_request(self, request, view):
if self.rate is None: # 要取init中看,self.rate=3/m,如果自己在频率类中写rate=5/m,我们就不需要写scope和配置文件了
return True
# get_cache_key返回了ip：192.168.1.19
self.key = self.get_cache_key(request, view)
if self.key is None: # 如果get_cache_key返回None，也不会做频率限制
return True
# self.history = self.VISIT_RECORD.get(self.key, [])
# self.cache 是缓存，相当于咱们的self.VISIT_RECORD
self.history = self.cache.get(self.key, [])
self.now = self.timer()# 取出当前时间
# self.duration：配置的1分钟访问3次，self.duration就是60
while self.history and self.history[-1] <= self.now - self.duration:
self.history.pop()
if len(self.history) >= self.num_requests: # '5/m',num_requests就是5
return self.throttle_failure()
return self.throttle_success()
# __init__
def __init__(self):
if not getattr(self, 'rate', None): # 如果没有，执行下面的代码
self.rate = self.get_rate()
self.num_requests, self.duration = self.parse_rate(self.rate)
# self.get_rate()
def get_rate(self):
# self.scope 我们自己写的字符串
# self.THROTTLE_RATES配置文件中的那个字典
return self.THROTTLE_RATES[self.scope] # 咱们配置的 3/m
# self.num_requests, self.duration = self.parse_rate(self.rate)---》self.rate是3/s
def parse_rate(self, rate):
if rate is None:
return (None, None)
# 3/m---->num=3,period=m
num, period = rate.split('/')
# num_requests=数字3
num_requests = int(num)
#
d={'s': 1, 'm': 60, 'h': 3600, 'd': 86400}
period[0]='m'
d[m]
duration = {'s': 1, 'm': 60, 'h': 3600, 'd': 86400}[period[0]]
return (num_requests, duration)