Django（64）频率认证源码分析与自定义频率认证

前言

有时候我们发送手机验证码，会发现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）
 

自定义频率认证

1. 自定义一个继承SimpleRateThrottle类的频率类
2. 设置一个scope类属性，属性值为任意见名知意的字符串
3. 在settings配置文件中，配置drf的DEFAULT_THROTTLE_RATES，格式为{scope对应的字符串值：'次数/时间'}
4. 在自定义频率类中重写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方法实现的