drf中仿照SimpleRateThrottle自定製頻率類
阿新 • • 發佈:2021-10-28
剛開始弄的時候不知道出現了什麼問題,計時出現負數,然後無法繼續訪問,後來也沒調整什麼就可以使用了
先放總程式碼
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]]
}, }
小白一個,學習筆記,還請指正