文章目录
- 一、频率组件
- 二、自定义频率类
- 1. 代码实现
- 2. 接口测试
一、频率组件
通过分析源码了解频率认证组件的方法调用过程
APIView 的 dispatch 中使用 initial 方法实现初始化并进行三大认证，第三步进行权限组件调用
rest_framework/views.py
class APIView(View):
# ...
# 定义默认频率类
throttle_classes = api_settings.DEFAULT_THROTTLE_CLASSES
def initial(self, request, *args, **kwargs):
# ...
# 认证组件：校验用户
# 这里调用 perform_authentication 实现认证
self.perform_authentication(request)
# 权限组件：校验用户权限
self.check_permissions(request)
# 频率组件：限制视图接口被访问次数
self.check_throttles(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())
# throttle_durations 非空，被加入等待时间，调用下面方法
if throttle_durations:
durations = [
duration for duration in throttle_durations
if duration is not None
]
duration = max(durations, default=None)
self.throttled(request, duration)
# 获取频率
def get_throttles(self):
# 由频率类 throttle_classes 定义
return [throttle() for throttle in self.throttle_classes]在 drf 设置文件查看默认权限配置
可以看到默认设置中并没有对访问频率做限制，也就是说可以无限次访问
rest_framework/settings.py
# 默认频率类配置
DEFAULTS = {
'DEFAULT_THROTTLE_CLASSES': [],
}详细看drf的默认频次类
settings.py
class BaseThrottle:
# 判断是否限次
# 无限次：返回True
# 限次：返回 False
def allow_request(self, request, view):
# 如果继承，必须重新
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
class SimpleRateThrottle(BaseThrottle):
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):
"""
Should return a unique cache-key which can be used for throttling.
Must be overridden.
May return `None` if the request should not be throttled.
"""
raise NotImplementedError('.get_cache_key() must be overridden')
def get_rate(self):
"""
Determine the string representation of the allowed request rate.
"""
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)
def parse_rate(self, rate):
"""
Given the request rate string, return a two tuple of:
<allowed number of requests>, <period of time in seconds>
"""
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)
def allow_request(self, request, view):
# 如果是 None，无限次访问
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()
def throttle_success(self):
# 访问成功，记录次数
self.history.insert(0, self.now)
self.cache.set(self.key, self.history, self.duration)
return True
def throttle_failure(self):
"""
Called when a request to the API has failed due to throttling.
"""
return False
def wait(self):
"""
Returns the recommended next request time in seconds.
"""
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)二、自定义频率类
1. 代码实现
- 继承 SimpleRateThrottle
- 设置 scope 类属性，属性值为任意见名知意的字符串
- 在 settings 配置中，配置drf的DEFAULT_THROTTLE_RATES，格式为 {scope: ‘次数/时间’}
- 在自定义频率类中重写 get_cache_key 方法
- 限制的对象返回：与限制信息有关的字符串
- 不限制的对象返回： None
- 实现根据自定义权限规则，确定是否有权限
- 进行全局或局部配置
- 全局：配置文件 settings.py
- 局部：在视图类 import
- 测试接口：前台在请求头携带认证信息，且默认规范用 Authorization 字段携带认证信息
throttles.py
from rest_framework.throttling import SimpleRateThrottle
# 短信频率限制
class SMSRateThrottle(SimpleRateThrottle):
scope = 'sms'
# 只对提交手机号的 get 方法进行限制
def get_cache_key(self, request, view):
mobile = request.query_params.get('mobile')
print(mobile)
# 没有手机号，不做频率限制
if not mobile:
return None
# 返回可以根据手机号动态变化，且不易重复的字符串，作为操作缓存的 key
return f'throttle_{self.scope}_{mobile}'settings.py
# 全局局部配置
REST_FRAMEWORK = {
# 配置频率限制条件
'DEFAULT_THROTTLE_RATES': {
'user': '3/min', # 用户 一分钟可访问三次
'anon': None, # 游客无限制
'sms': '1/min' # sms 一分钟可访问一次
}
}views.py
from rest_framework.views import APIView
from api.throttles import SMSRateThrottle
from utils.response import APIResponse
class SMSAPIView(APIView):
throttle_classes = [SMSRateThrottle]
def get(self, request, *args, **kwargs):
return APIResponse(0, 'Verification code successful')
def post(self, request, *args, **kwargs):
return APIResponse(0, 'Verification code successful')urls.py
from django.conf.urls import url
from api import views
urlpatterns = [
url(r'^sms/$', views.SMSAPIView.as_view()),
]2. 接口测试
第一次调用
调用一次后，第二次返回
