大家好,又见面了,我是你们的朋友全栈君。如果您正在找激活码,请点击查看最新教程,关注关注公众号 “全栈程序员社区” 获取激活教程,可能之前旧版本教程已经失效.最新Idea2022.1教程亲测有效,一键激活。
Jetbrains全系列IDE使用 1年只要46元 售后保障 童叟无欺
前言
有时候我们发送手机验证码,会发现1分钟只能发送1次,这是做了频率限制,限制的时间次数,都由开发者自己决定
频率认证源码分析
def check_throttles(self, request):
"""
检查是否应限制请求。如果请求受到限制,则引发适当的异常。
"""
throttle_durations = []
# 1.遍历配置的频率认证类,初始化一个个频率认证类对象(会调用频率认证类的__init__()方法)
# 2.频率认证类对象调用allow_request()方法,频率是否限次(没有限次可访问,限次不可访问)
# 3.频率认证类限次后,调用wait方法,获取还需多长时间可以进行下一次访问
for throttle in self.get_throttles():
if not throttle.allow_request(request, self):
throttle_durations.append(throttle.wait())
if throttle_durations:
# Filter out `None` values which may happen in case of config / rate
# changes, see #1438
durations = [
duration for duration in throttle_durations
if duration is not None
]
duration = max(durations, default=None)
self.throttled(request, duration)
get_throttles()
我们首先来查看get_throttles()
源码
def get_throttles(self):
"""
实例化并返回此视图使用的节流阀列表。
"""
return [throttle() for throttle in self.throttle_classes]
然后点击throttle_classes
,跳转到APIView
后查看源码
throttle_classes = api_settings.DEFAULT_THROTTLE_CLASSES
接着我们去settings.py
文件中查看,发现'DEFAULT_THROTTLE_CLASSES': [],
默认是一个空列表,那么我们就知道了for throttle in self.get_throttles()
其实是去遍历列表中配置的频率认证,至于列表中需要填写什么,我们后续再看
allow_request
接下来我们查看allow_request
方法,它是drf
中的throtting.py
文件中BaseThrottle
类中的方法,我们查看下BaseThrottle
源码
class BaseThrottle:
"""
Rate throttling of requests.
"""
def allow_request(self, request, view):
"""
如果应该允许请求,则返回 `True`,否则返回 `False`。
"""
raise NotImplementedError('.allow_request() must be overridden')
def get_ident(self, request):
"""
Identify the machine making the request by parsing HTTP_X_FORWARDED_FOR
if present and number of proxies is > 0. If not use all of
HTTP_X_FORWARDED_FOR if it is available, if not use REMOTE_ADDR.
"""
xff = request.META.get('HTTP_X_FORWARDED_FOR')
remote_addr = request.META.get('REMOTE_ADDR')
num_proxies = api_settings.NUM_PROXIES
if num_proxies is not None:
if num_proxies == 0 or xff is None:
return remote_addr
addrs = xff.split(',')
client_addr = addrs[-min(num_proxies, len(addrs))]
return client_addr.strip()
return ''.join(xff.split()) if xff else remote_addr
def wait(self):
"""
返回推荐的在下一个请求之前等待的秒数
"""
return None
可以看到BaseThrottle
类下有3个方法
allow_request
:如果需要继承该类,必须重写此方法get_ident
:获取身份wait
:返回等待的秒数
SimpleRateThrottle
而throtting
中有个SimpleRateThrottle
继承自BaseThrottle
,我们大多数情况下都会自定义SimpleRateThrottle
类,让我们查看下源码,看他干了哪些事情
class SimpleRateThrottle(BaseThrottle):
"""
一个简单的缓存实现,只需要提供get_cache_key方法即可
速率(requests / seconds)由 View 类上的 `rate` 属性设置。该属性是“number_of_requests/period”形式的字符串。
period应该是以下之一:('s', 'sec', 'm', 'min', 'h', 'hour', 'd', 'day')
用于限制的先前请求信息存储在缓存中
"""
cache = default_cache
timer = time.time
cache_format = 'throttle_%(scope)s_%(ident)s'
scope = None
THROTTLE_RATES = api_settings.DEFAULT_THROTTLE_RATES
def __init__(self):
if not getattr(self, 'rate', None):
self.rate = self.get_rate()
self.num_requests, self.duration = self.parse_rate(self.rate)
def get_cache_key(self, request, view):
def get_rate(self):
def parse_rate(self, rate):
def allow_request(self, request, view):
def throttle_success(self):
def throttle_failure(self):
def wait(self):
我们可以看到SimpleRateThrottle
有5个属性
cache
:默认的django
中的缓存timer
:当前时间cache_format
:缓存的格式throttle_%(scope)s_%(ident)s
scope
:范围THROTTLE_RATES
:默认的频率
除了属性,还有8个方法,我们依次查看源码
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)
代码讲解:如果没有获取到rate
属性,那么rate
属性就从get_rate()
方法中获取,拿到后,从parse_rate
方法中解析出一个元组,包含2个元素num_requests
和duration
num_request
:请求次数duration
:持续时间
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)
代码讲解:如果没有获取到scope
属性,会抛出异常信息,如果有scope
就从THROTTLE_RATES[self.scope]
中返回它,THROTTLE_RATES
默认值如下:
'DEFAULT_THROTTLE_RATES': {
'user': None,
'anon': None,
},
所以get_rate
方法返回的是THROTTLE_RATES
中key
为scope
所对应的值,scope
属性我们可以自定义的时候随意设置,如果我们自定义scope
为user
,那么get_rate
方法返回的就是None
,所以self.rate
也就为None
parse_rate
获取到rate
,用此方法解析
def parse_rate(self, rate):
"""
提供请求速率字符串,返回一个二元组
允许请求的次数, 以秒为单位的时间段
"""
if rate is None:
return (None, None)
num, period = rate.split('/')
num_requests = int(num)
duration = {'s': 1, 'm': 60, 'h': 3600, 'd': 86400}[period[0]]
return (num_requests, duration)
代码讲解:如果rate
为None
,那么就返回(None, None)
,如果不为None
,会把rate
以/
分割,从这里我们就知道了rate
的字符串的形式就是num/period
,比如3/min
,最终会把他分割,然后返回一个元组
- num_requests:请求的次数
- duration:取
period
中的下标为0的,然后从字典中取出对应的key
的值,比如min
,第一个开头字母为m
,最后从字典中取m
的值,就是60
所以示例3/min
代表的就是1分钟可以访问3次
get_cache_key
def get_cache_key(self, request, view):
"""
应该返回可用于限制的唯一cache-key。必须被覆盖。
如果不限制请求,则可能返回“None”。
"""
raise NotImplementedError('.get_cache_key() must be overridden')
这个方法很简单,就是获取唯一的缓存key
,如果请求不做限制,则返回None
allow_request
由于父类BaseThrottle
的allow_request
方法没有实现具体的逻辑,所以SimpleRateThrottle
中实现了具体的细节
def allow_request(self, request, view):
"""
如果请求应该被节流,那么实行检查以便查看
成功时调用`throttle_success`.
失败时调用`throttle_failure`.
"""
if self.rate is None:
return True
self.key = self.get_cache_key(request, view)
if self.key is None:
return True
self.history = self.cache.get(self.key, [])
self.now = self.timer()
# 从历史记录中删除现在已经超过节流持续时间的任何请求
while self.history and self.history[-1] <= self.now - self.duration:
self.history.pop()
if len(self.history) >= self.num_requests:
return self.throttle_failure()
return self.throttle_success()
代码讲解:如果rate
为None
就返回True
,代表允许请求,如果key
也为None
则返回True
,代表允许请求,如果rate
和key
都有值,history
就从缓存中获取key
所对应的列表,now
代表当前时间。如果history
有值,并且列表history
的最后一个元素≤当前时间-持续时间,那么history
列表就会删除这个元素,如果列表长度≥请求次数,就会调用throttle_failure
,如果列表长度<请求次数,则调用throttle_success
。
举例:如果self.now
假设为晚上20:00,duration
和num_requests
就用之前3/min
的示例,duration
表示60s,num_requests
表示3次,那么self.now-self.duration
就代表19:59分,如果history
列表中的最后一个元素的时间值≤19:59,那么就删除它,我们的需求是3/min
一分钟只能访问3次,而你超过了1分钟,就没必要限制了,所以将时间从history
删除,如果history
列表长度≥3,一开始是空列表的时候不满足条件,会返回throttle_success
,第二次访问列表长度会增加到1,但还是不满足条件,会继续调用throttle_success
,第三次访问列表长度为2,仍然不满足会继续调用throttle_success
,第四次访问满足条件,就会调用throttle_failure
,代表不能再请求了
throttle_success
def throttle_success(self):
"""
将当前请求的时间戳与键一起插入缓存中。
"""
self.history.insert(0, self.now)
self.cache.set(self.key, self.history, self.duration)
return True
代码详解:将当前时间插入到history
列表的头部,给缓存设置key
的值为当前时间,超时时间为duration
,最后返回True
,代表可以访问
throttle_failure
def throttle_failure(self):
"""
当对 API 的请求由于节流而失败时调用。
"""
return False
返回False
,代表请求节流失败,不允许访问
wait
def wait(self):
"""
以秒为单位返回推荐的下一个请求时间。
"""
if self.history:
remaining_duration = self.duration - (self.now - self.history[-1])
else:
remaining_duration = self.duration
available_requests = self.num_requests - len(self.history) + 1
if available_requests <= 0:
return None
return remaining_duration / float(available_requests)
代码解析:如果history
列表存在,remaining_duration
剩余时间就等于持续时间减去(当期时间-列表最后一个元素的时间),如果self.now
为晚上20:00,history
的最后一个元素值为19:59:30
,而持续时间duration
设置为60s,那么remaining_duration
就代表还剩30s就可以进行访问了,而available_requests
可用请求等于(设置好的请求次数-history
列表+1)
自定义频率认证
- 自定义一个继承
SimpleRateThrottle
类的频率类 - 设置一个
scope
类属性,属性值为任意见名知意的字符串 - 在
settings
配置文件中,配置drf
的DEFAULT_THROTTLE_RATES
,格式为{scope
对应的字符串值:’次数/时间’} - 在自定义频率类中重写
get_cache_key
方法
限制的对象返回与限制信息有关的字符串
不限制的对象返回None
需求:用户访问短信验证码1分钟只能发送1次验证码
我们创建一个throttles.py
文件,然后定义SMSRateThrottle
类,代码如下:
from rest_framework.throttling import SimpleRateThrottle
class SMSRateThrottle(SimpleRateThrottle):
scope = "sms"
def get_cache_key(self, request, view):
phone = request.query_params.get('phone') or request.data.get('phone')
# 没有手机号,就不做频率限制
if not phone:
return None
# 返回可以根据手机号动态变化,且不易重复的字符串,作为操作缓存的key
return f"throttle_{self.scope}_{phone}"
在settings.py
文件中配置DEFAULT_THROTTLE_RATES
,代码如下:
'DEFAULT_THROTTLE_RATES': {
'sms': '1/min'
},
最后再视图函数中,局部配置自定义认证类
class TestView(APIView):
throttle_classes = [SMSRateThrottle]
def get(self, request, *args, **kwargs):
return APIResponse(data_msg="get 获取验证码")
def post(self, request, *args, **kwargs):
return APIResponse(data_msg="post 获取验证码")
具体测试细节过程就不再描述了,这里只讲述结果,当我们使用get
或者post
请求时,携带请求参数phone
第一次发送请求,请求成功,第二次就会出现以下提示
{
"detail": "请求超过了限速。 Expected available in 58 seconds."
}
58 seconds
代表还剩58秒可以再次访问,至于58s是怎么算出来的,就是SimpleRateThrottle
类中的wait
方法实现的
发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/164877.html原文链接:https://javaforall.cn
【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛
【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...