1. 程式人生 > 其它 >drf中仿照SimpleRateThrottle自定製頻率類

drf中仿照SimpleRateThrottle自定製頻率類


剛開始弄的時候不知道出現了什麼問題,計時出現負數,然後無法繼續訪問,後來也沒調整什麼就可以使用了

先放總程式碼

from django.core.exceptions import ImproperlyConfigured

from rest_framework.throttling import SimpleRateThrottle
from rest_framework.throttling import BaseThrottle
import time

from drf_book import settings

class
IPThrottle(BaseThrottle): # 定義成類屬性所有物件都用這一個
VISIT_DIC = {} # 儲存來訪問的IP, 個人測試使用 若是用在伺服器上應該存到快取裡面,而不是存在伺服器上 scope = 'ming' # 用來查詢相應頻率 THROTTLE_RATES = settings.REST_FRAMEWORK['DEFAULT_THROTTLE_RATES'] def __init__(self): self.history_list = [] self.rate = self.get_rate() self.num_request, self.duration
= self.parse_rate(self.rate) def allow_request(self, request, view): """ 1首先取出Ip地址 2、判斷當前Ip不在訪問字典中,新增進去,並且直接返回True, 表示第一次訪問 3、迴圈判斷當前的ip的列表, 有值, 並且 當前時間減去列表的最後一個時間大於設定的時間, 把這種資料pop掉,這樣列表中只有指定時間以內的訪問時間,直到列表清空,這時候就相當於首次訪問 4、判斷,當列表小於指定次數, 把當前時間插入到列表第一個位置,返回True順利通過 5、當大於等於指定次數, 說明一分鐘之內訪問超過3次, 返回False驗證失敗
""" ip = request.META.get('REMOTE_ADDR') # 獲取訪問者的IP ctime = time.time() if ip not in self.VISIT_DIC: self.VISIT_DIC[ip] = [ctime, ] return True self.history_list = self.VISIT_DIC[ip] while True: if self.history_list and ctime - self.history_list[-1] > self.duration: # 要注意可能存在列表被清空的情況 self.history_list.pop() # 把最後一個去掉 else: break if len(self.history_list) < self.num_request: self.history_list.insert(0, ctime) # 插入到第一個位置 return True else: return False 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) else: num, period = rate.split('/') num_request = int(num) duration = {'s': 1, 'm': 60, 'h': 3600, 'd': 86400}[period[0]] return (num_request, duration) def wait(self): """ Returns the recommended next request time in seconds. """ # 當前時間,減去列表中最後一個時間 ctime = time.time() return self.duration - (ctime - self.history_list[-1])

注意:

  • 首先要明白在頻率限制中真正起作用的是一個屬性allow_request(self, request, view),自定製的類可以繼承BaseThrottle,也可以不繼承
  • 所以要自定製,就要重寫allow_request(self, request, view),看一下BaseThrottle中的,看一下注釋,意思就是允許訪問返回Ture,反之False
 def allow_request(self, request, view):
        """
        Return `True` if the request should be allowed, `False` otherwise.
        """
        raise NotImplementedError('.allow_request() must be overridden')
  • 函式wait的返回值是等待的時間
    def wait(self):
        """
        Optionally, return a recommended number of seconds to wait before
        the next request.
        """
        return None
  • parse_rate和get_rate都是用來獲取配置檔案中的頻率的,配置檔案中的配置如下
REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_CLASSES': ['utiils.throttling.IPThrottle', ],  # 自定製頻率類的路徑
    'DEFAULT_THROTTLE_RATES': {
            'ming': '3/m', # duration = {'s': 1, 'm': 60, 'h': 3600, 'd': 86400}[period[0]]
}, }

小白一個,學習筆記,還請指正