用戶訪問頻率控制原始碼剖析，和用戶登錄驗證有點相似，但是為了增加記憶，有必要再一次添加，

注意：一定要跟著博主的解說再看代碼的中文注釋及其下面的一行代碼！！！

1、準備一個路由和視圖類，全域路由配置暫時忽略，當流程執行到下面的url：groupsSelectAll——> GroupsView的視圖類下的as_view()方法

from django.conf.urls import url

from . import views


app_name = '[words]'
urlpatterns = [
    url(r'groupsSelectAll/', views.GroupsView.as_view(), name="groupsSelectAll"),   # 詞組資訊查詢所有

]

class GroupsView(APIView):

    def get(self, request):
        conditions = {
            "id": request.query_params.get("wid"),
            "name": request.query_params.get("name"),
            "start_time": request.query_params.get("start_time"),
            "end_time": request.query_params.get("end_time"),
        }
        res = DataManager.select_by_conditions("words_groups", None, **conditions)
        return Response(data=https://www.cnblogs.com/aitiknowledge/p/{"code": 200, "result": res})

2、但是GroupsView類下沒有as_view方法，這時就要去它的父類APIView查看（點進去看as_view方法），這里博主只復制方法源代碼，大家只需要看中文注釋及其下的代碼陳述句，在這個方法中值得一提的是super關鍵字，如果請求視圖類（就是GroupsView類，如果繼承了多個父類）還有另一個父類，它先會查看這個父類是否有as_view方法，在這里它是會執行APIView的父類View中的as_view方法，然后我們再次查看父類View的as_view方法，第一個as_view方法是APIView類的，第二個as_view方法是View類的，

@classmethod
    def as_view(cls, **initkwargs):
        """
        Store the original class on the view function.

        This allows us to discover information about the view when we do URL
        reverse lookups.  Used for breadcrumb generation.
        """
        if isinstance(getattr(cls, 'queryset', None), models.query.QuerySet):
            def force_evaluation():
                raise RuntimeError(
                    'Do not evaluate the `.queryset` attribute directly, '
                    'as the result will be cached and reused between requests. '
                    'Use `.all()` or call `.get_queryset()` instead.'
                )
            cls.queryset._fetch_all = force_evaluation

        # 執行父類的as_view方法
        view = super(APIView, cls).as_view(**initkwargs)
        view.cls = cls
        view.initkwargs = initkwargs

        # Note: session based authentication is explicitly CSRF validated,
        # all other authentication is CSRF exempt.
        return csrf_exempt(view)

    @classonlymethod
    def as_view(cls, **initkwargs):
        """Main entry point for a request-response process."""
        for key in initkwargs:
            if key in cls.http_method_names:
                raise TypeError("You tried to pass in the %s method name as a "
                                "keyword argument to %s(). Don't do that."
                                % (key, cls.__name__))
            if not hasattr(cls, key):
                raise TypeError("%s() received an invalid keyword %r. as_view "
                                "only accepts arguments that are already "
                                "attributes of the class." % (cls.__name__, key))

        # 執行view方法
        def view(request, *args, **kwargs):
            # 這里的cls就是我們的請求視圖類，顯而易見，這個self是請求試圖類的物件
            self = cls(**initkwargs)
            if hasattr(self, 'get') and not hasattr(self, 'head'):
                self.head = self.get
            self.request = request
            self.args = args
            self.kwargs = kwargs
            # 然后這里就是執行dispatch方法
            return self.dispatch(request, *args, **kwargs)
        view.view_class = cls
        view.view_initkwargs = initkwargs

        # take name and docstring from class
        update_wrapper(view, cls, updated=())

        # and possible attributes set by decorators
        # like csrf_exempt from dispatch
        update_wrapper(view, cls.dispatch, assigned=())
        return view

3、我們在第二個as_view方法中可以知道self是我們的請求視圖類的物件，通過這個self呼叫dispatch方法，請求視圖類中沒有dispatch方法，是不是又去APIView類中執行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
        # 這里是對原生的request加工處理，回傳一個新的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
            if request.method.lower() in self.http_method_names:
                # 通過python的反射機制反射到請求視圖類的方法名稱
                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

4、其他代碼不用看，我們直接看initial方法，因為這個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
        # 用戶驗證的方法，這個request 是加工之后的request
        self.perform_authentication(request)
        # 用戶權限驗證
        self.check_permissions(request)
        # 用戶訪問頻率限制
        self.check_throttles(request)

5、這就到了我們的用戶訪問頻率控制的戲碼了，博主添加APIView部分代碼，即check_throttles方法用到的代碼，我們可以查看代碼中的self.check_chrottles(request)，點進去查看check_chrottles()方法，可以看到有get_throttles方法，這個方法有self.throttles_classes變數，即self.throttles_classes = api_settings.DEFAULT_THROTTLES_CLASSES，然后這里也和【上一篇的用戶權限驗證】很相似，就是請求視圖類中如果沒有這個變數名及值（值是一個串列），就會使用全域組態檔中的REST_FRAMEWORK={"DEFAULT_THROTTLES_CLASSES":["訪問頻率控制類的全路徑"]}，或者我們在請求視圖類中添加這個變數及值

class APIView(View):

    # 如果請求視圖類中沒有這個變數和值，就會使用全域組態檔的值
    throttle_classes = api_settings.DEFAULT_THROTTLE_CLASSES

    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():
            # 執行物件下的方法allow_request(),如果為True就不做處理，如果為False，就拋出例外資訊
            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]

    def throttled(self, request, wait):
        """
        If request is throttled, determine what kind of exception to raise.
        """
        # 拋出訪問頻率控制類的例外
        raise exceptions.Throttled(wait)

6、在上面的APIView類中會執行到if not throttle.allow_throttles(request, self),我們可以直接點進去查看allow_request方法，進入有關于Throttles的類：BaseThrottle類主要提供了三個方法，但一般不會繼承這個類，而是繼承SimpleRateThrottle類，我們看看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.
    """
    # Django默認的快取
    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):
        # 尋找繼承該類的方法是否含有rate方法或者變數名，如果沒有，就執行get_rate函式
        if not getattr(self, 'rate', None):
            # 拿到的值是"3/m"
            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.
        """
        # 通過反射機制呼叫子類或者當前類的scope函式或者變數名，如果有的話就執行try里面的陳述句
        if not getattr(self, 'scope', None):
            msg = ("You must set either `.scope` or `.rate` for '%s' throttle" %
                   self.__class__.__name__)
            raise ImproperlyConfigured(msg)

        try:
            # 獲取配置型別的scope變數或者方法，這里應該回傳一個比例
            # 在組態檔中假設一個值為 "3/m"
            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>
        """
        # rate = "3/m"
        if rate is None:
            return (None, None)
        num, period = rate.split('/')
        # num_request = 3
        num_requests = int(num)
        # duration = 60
        duration = {'s': 1, 'm': 60, 'h': 3600, 'd': 86400}[period[0]]
        # (3, 60)
        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)

這時我們自定義訪問頻率控制類時，繼承這個類，撰寫就很簡單了

class MyThrottle(SimpleRateThrottle):
    # 這個就是在全域組態檔上的鍵
    scope = "my_rate"

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


# 在組態檔上的配置
REST_FRAMEWORK = {
    "DEFAULT_THROTTLE_CLASSES":["MyThrottle的全路徑"],
    "DEFAULT_THROTTLE_RATE:{
        "my_rate": "3/m"
    }
}

標籤：Python

上一篇：隨機生成簡單驗證碼

下一篇：python基礎（二）