Django REST 框架详解 10 | 频率认证组件

一、频率组件

通过分析源码了解频率认证组件的方法调用过程

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 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 = {
    # 默认频率类配置
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 / 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}_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' # 全局局部配置
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, 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(from django.conf.urls import url
from api import views

urlpatterns = [
    url(r'^sms/$', views.SMSAPIView.as_view()),
]

2. 接口测试

第一次调用

调用一次后，第二次返回