rest framework之限流组件

Posted 2022-10-13 shenjianping

一、自定义限流

限流组件又叫做频率组件，用于控制客户端可以对API进行的请求频率，比如说1分钟访问3次，如果在1分钟内超过3次就对客户端进行限制。

1、自定义限流

假设现在对一个API访问，在30s内访问不能超过3次，应该如何实现？

VISIT_RECORD =  #定义全局变量，用于存放访问记录
class VisitThrottle(object):

    def __init__(self):

　　　　 #用于await计算剩余访问时间
        self.history = None

    def allow_request(self,request,view):
        #获取用户ip作为唯一的标示
        remote_addr = request.META.get(‘REMOTE_ADDR‘)

        # 获取当前访问的时刻
        ctime = time.time()
        # 这是用户第一次访问,将其进行记录，并且返回True，允许继续访问
         if remote_addr not in VISIT_RECORD:
             VISIT_RECORD[remote_addr] = [ctime,]
             return True

        # 如果不是第一次访问，获取所有的记录
        history = VISIT_RECORD.get(remote_addr)

        self.history = history
        # 判断最开始的时刻与现在的时刻的差值是否在规定的时间范围内，比如在60s内，如果不在，
        # 可以去除最开始的时刻记录
        while history and history[-1] < ctime - 30:
            history.pop()
        # 此时列表中的时刻记录都是在规定的时间范围内，判断时刻的个数也就是访问的次数

        if len(history) < 3:
            history.insert(0,ctime)
            return True

    def wait(self):
        # 还需要等多少秒才能访问
        ctime = time.time()
        return 60 - (ctime - self.history[-1])

在对应的视图中进行配置：

class BookView(ListAPIView):
    throttle_classes = [VisitThrottle,] #配置限流组件
    queryset = models.Book.objects.all()
    serializer_class = BookModelSerializer

2、限流原理

　　在rest framework框架中，限流定义为类的列表，只需要全局配置或者局部配置即可。上述限流的原理就是以客户端的唯一标示作为键，以访问的时刻形成的列表作为值形成的字典，然后通过对字典进行操作：

    http://127.0.0.1:8020/ :[11:43:30,11:42:22,11:42:20,11:42:09]

　　如上面的字典所示，后面的访问时间放插入到列表的最左侧，加入当前访问时间是11:43::30，那么与最开始访问时间11:42:09进行做差，然后与规定时间30s进行比较，如果不在30s内，那么就去除最左边的记录，同理使用while循环依次比较，最后在规定时间范围内的记录：

　　http://127.0.0.1:8020/ :[11:43:30,]

再计算访问次数，也就是列表的个数，显然如果列表的个数小于3可以继续访问，否则不可以。

　　上面使用全局变量来进行记录，当然也是可以使用缓存来进行记录的存储，需要使用django的缓存API，from django.core.cache import cache，导入这个API后就可以使用set和get方法，设置和获取cache中存储的对象，只需要在操作全局变量除进行替换即可：

from django.core.cache import cache as default_cache
import time

class VisitThrottle(object):

    cache = default_cache
     
    def allow_request(self,request,view):
         ...
         ...    
        # 这是用户第一次访问,将其进行记录，并且返回True，允许继续访问

        if not self.cache.get(remote_addr):
            self.cache.set(remote_addr,[ctime,])
            return True
        # 如果不是第一次访问，获取所有的记录

        history = self.cache.get(remote_addr)
        self.history = history
        ...
        ...

rest framework的限流组件就是基于cache来完成的。　　

　　上述的wait方法表示还需要多长时间可以进行访问这个API，对客户端的提示：

    "detail": "Request was throttled. Expected available in 56 seconds."

二、内置限流

在rest framework中已经有一些限流的API可以使用：

1、SimpleRateThrottle

class SimpleRateThrottle(BaseThrottle):
    """
    A simple cache implementation, that only requires `.get_cache_key()`
    to be overridden.

    The rate (requests / seconds) is set by a `rate` attribute on the View
    class.  The attribute is a string of the form ‘number_of_requests/period‘.

    Period should be one of: (‘s‘, ‘sec‘, ‘m‘, ‘min‘, ‘h‘, ‘hour‘, ‘d‘, ‘day‘)

    Previous request information used for throttling is stored in the cache.
    """
    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):
        """
        Implement the check to see if the request should be throttled.

        On success calls `throttle_success`.
        On failure calls `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()

        # Drop any requests from the history which have now passed the
        # throttle duration
        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):
        """
        Inserts the current request‘s timestamp along with the key
        into the cache.
        """
        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)

SimpleRateThrottle

如果需要借助这个API来实现功能，自己也需要进行一些配置：

• 继承SimpleRateThrottle

自己定义的限流类需要继承SimpleRateThrottle：

class VisitThrottle(SimpleRateThrottle):
    ...

• 设置scope

class VisitThrottle(SimpleRateThrottle):
    scope = ‘book‘
    ...

在自定义类中设置scope参数，并且还需要在settings中配置DEFAULT_THROTTLE_RATES

REST_FRAMEWORK = 

    "DEFAULT_THROTTLE_RATES": 
        "book": ‘6/m‘,  #每分钟访问6次

    

• 重写get_cache_key方法

class VisitThrottle(SimpleRateThrottle):
    scope = ‘book‘

    def get_cache_key(self,request,view):
        """
        获取访问的标示，比如以ip作为标示
        :param request:
        :param view:
        :return:
        """
        remote_addr = request.META.get(‘REMOTE_ADDR‘)
        return remote_addr

获取访问的唯一标示ip，当然SimpleRateThrottle继承了BaseThrottle，在BaseThrottle中有获取ip的方法，只需要调用即可。

class VisitThrottle(SimpleRateThrottle):
    scope = ‘book‘

    def get_cache_key(self,request,view):
        return self.get_ident(request)

• 局部配置

只需要在对应的视图中添加对应限流类的列表即可：

class BookView(ListAPIView):
    ...
    throttle_classes = [VisitThrottle,] #配置节流组件
    ...

• 全局配置

当然也可以在settings中进行全局配置：

REST_FRAMEWORK = 

"DEFAULT_THROTTLE_CLASSES":["app01.utils.throttle.VisitThrottle"],

    "DEFAULT_THROTTLE_RATES": 
        "book": ‘6/m‘,

    

这样也就完成了相对应的功能，另外内部还提供了其它的API可以使用。

2、AnonRateThrottle

class AnonRateThrottle(SimpleRateThrottle):
    """
    Limits the rate of API calls that may be made by a anonymous users.

    The IP address of the request will be used as the unique cache key.
    """
    scope = ‘anon‘

    def get_cache_key(self, request, view):
        if request.user.is_authenticated:
            return None  # Only throttle unauthenticated requests.

        return self.cache_format % 
            ‘scope‘: self.scope,
            ‘ident‘: self.get_ident(request)
        

AnonRateThrottle

 限制未认证的用户。通过传入请求的 IP 地址生成一个唯一的密钥来进行限制。

3、UserRateThrottle

class UserRateThrottle(SimpleRateThrottle):
    """
    Limits the rate of API calls that may be made by a given user.

    The user id will be used as a unique cache key if the user is
    authenticated.  For anonymous requests, the IP address of the request will
    be used.
    """
    scope = ‘user‘

    def get_cache_key(self, request, view):
        if request.user.is_authenticated:
            ident = request.user.pk
        else:
            ident = self.get_ident(request)

        return self.cache_format % 
            ‘scope‘: self.scope,
            ‘ident‘: ident
        

UserRateThrottle

　　通过 API 将用户请求限制为给定的请求频率。用户标识用于生成一个唯一的密钥来加以限制。未经身份验证的请求将回退到使用传入请求的 IP 地址生成一个唯一的密钥来进行

限制。

4、ScopedRateThrottle

class ScopedRateThrottle(SimpleRateThrottle):
    """
    Limits the rate of API calls by different amounts for various parts of
    the API.  Any view that has the `throttle_scope` property set will be
    throttled.  The unique cache key will be generated by concatenating the
    user id of the request, and the scope of the view being accessed.
    """
    scope_attr = ‘throttle_scope‘

    def __init__(self):
        # Override the usual SimpleRateThrottle, because we can‘t determine
        # the rate until called by the view.
        pass

    def allow_request(self, request, view):
        # We can only determine the scope once we‘re called by the view.
        self.scope = getattr(view, self.scope_attr, None)

        # If a view does not have a `throttle_scope` always allow the request
        if not self.scope:
            return True

        # Determine the allowed request rate as we normally would during
        # the `__init__` call.
        self.rate = self.get_rate()
        self.num_requests, self.duration = self.parse_rate(self.rate)

        # We can now proceed as normal.
        return super(ScopedRateThrottle, self).allow_request(request, view)

    def get_cache_key(self, request, view):
        """
        If `view.throttle_scope` is not set, don‘t apply this throttle.

        Otherwise generate the unique cache key by concatenating the user id
        with the ‘.throttle_scope` property of the view.
        """
        if request.user.is_authenticated:
            ident = request.user.pk
        else:
            ident = self.get_ident(request)

        return self.cache_format % 
            ‘scope‘: self.scope,
            ‘ident‘: ident
        

ScopedRateThrottle

　　可用于限制对 API 特定部分的访问。只有当正在访问的视图包含 .throttle_scope 属性时才会应用此限制。然后通过将请求的 “范围” 与唯一的用户标识或 IP 地址连接起来形成唯一的限流密钥

三、源码剖析

限流组件和权限组件、认证组件等类似，还是从路由对应的视图函数的as_view方法着手，可以看到最终走到的还是APIView的dispatch方法。

1、dispatch

   def dispatch(self, request, *args, **kwargs):
        """
        `.dispatch()` is pretty much the same as Django‘s regular dispatch,
        but with extra hooks for startup, finalize, and exception handling.
        """
        self.args = args
        self.kwargs = kwargs
        #rest-framework重构request对象
        request = self.initialize_request(request, *args, **kwargs)
        self.request = request
        self.headers = self.default_response_headers  # deprecate?
        try:
            self.initial(request, *args, **kwargs)

            # Get the appropriate handler method
            #这里和CBV一样进行方法的分发
            if request.method.lower() in self.http_method_names:
                handler = getattr(self, request.method.lower(),
                                  self.http_method_not_allowed)
            else:
                handler = self.http_method_not_allowed

            response = handler(request, *args, **kwargs)

        except Exception as exc:
            response = self.handle_exception(exc)

        self.response = self.finalize_response(request, response, *args, **kwargs)
        return self.response

这里的dispatch方法是APIView中的dispatch方法，在这里对原先的request进行了重构，以及通过self.initial(request, *args, **kwargs)加入了限流组件。

2、initial

    def initial(self, request, *args, **kwargs):
        """
        Runs anything that needs to occur prior to calling the method handler.
        """
        self.format_kwarg = self.get_format_suffix(**kwargs)

        # Perform content negotiation and store the accepted info on the request
        neg = self.perform_content_negotiation(request)
        request.accepted_renderer, request.accepted_media_type = neg

        # Determine the API version, if versioning is in use.
        version, scheme = self.determine_version(request, *args, **kwargs)
        request.version, request.versioning_scheme = version, scheme

        # Ensure that the incoming request is permitted
        self.perform_authentication(request) #进行认证
        self.check_permissions(request) #权限校验
        self.check_throttles(request) #限流组件

3、check_throttles

    def check_throttles(self, request):
        """
        Check if request should be throttled.
        Raises an appropriate exception if the request is throttled.
        """
        for throttle in self.get_throttles():
            if not throttle.allow_request(request, self):
                self.throttled(request, throttle.wait())

    def get_throttles(self):
        """
        Instantiates and returns the list of throttles that this view uses.
        """
        return [throttle() for throttle in self.throttle_classes]

get_throttles

　　可以看到循环的是视图中配置的限流类的列表，而且显然每一个限流类都必须要有allow_request和wait方法，如果allow_request返回的False就是说明已经限制访问了，执行self.throttled(request, throttle.wait())。

4、throttled

    def throttled(self, request, wait):
        """
        If request is throttled, determine what kind of exception to raise.
        """
        raise exceptions.Throttled(wait)

也就是如果已经限流了，就会抛出异常，给客户端限流提示。

详情参考：https://q1mi.github.io/Django-REST-framework-documentation/api-guide/throttling/#anonratethrottle