drf之頻率類原始碼

2023-06-01 21:00:52

1 頻率類

寫一個類,繼承SimpleRateThrottle,重寫get_cache_key,返回[ip,使用者id]什麼,就以什麼做限制,編寫類屬性  scope = 字串,在組態檔中設定
      'DEFAULT_THROTTLE_RATES': {
        '字串': '3/m',
      }
      設定在檢視類,全域性使用(設定在組態檔中)

2 自定義頻率類

    原始碼中找---》自定義頻率類如何寫---》寫個類,重寫allow_request,如果能存取就返回True,不能存取返回False

    APIView---》dispatch---》走三大認證的時候
    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())

        if throttle_durations:
            durations = [
               duration for duration in throttle_durations
               if duration is not None
            ]
            duration = max(durations, default=None)
            self.throttled(request, duration)

# 3 自定義頻率類 這是我們自己寫的
class MyThrottles():
    VISIT_RECORD = {} # 存取者 ip 字典,格式是{ip1:[時間4,時間3,時間2,時間1],ip2:[時間,時間],ip3:[當前時間]}
    def __init__(self):
        self.history = None
    def allow_request(self, request, view):
        # (1)取出存取者ip
        # print(request.META)
        ip = request.META.get('REMOTE_ADDR')
        import time
        ctime = time.time()
        # (2)判斷當前ip不在存取字典裡,新增進去,並且直接返回True,表示第一次存取
        if ip not in self.VISIT_RECORD:
            # {ip地址作為key:[當前時間,時間1,時間2,時間3]}
            self.VISIT_RECORD[ip] = [ctime, ]
            return True
        self.history = self.VISIT_RECORD.get(ip)
        # (3)迴圈判斷當前ip的列表,有值,並且當前時間減去列表的最後一個時間大於60s,把這種資料pop掉,這樣列表中只有60s以內的存取時間,
        while self.history and ctime - self.history[-1] > 60:
            self.history.pop()
        # (4)判斷,當列表小於3,說明一分鐘以記憶體取不足三次,把當前時間插入到列表第一個位置,返回True,順利通過
        # (5)當大於等於3,說明一分鐘記憶體取超過三次,返回False驗證失敗
        if len(self.history) < 3:
            self.history.insert(0, ctime)
            return True
        else:
            return False
    def wait(self):
        import time
        ctime = time.time()
        return 60 - (ctime - self.history[-1])

SimpleRateThrottle原始碼分析

    """
    -allow_request:必須有的,頻率類,必須重寫它
    -get_cache_key:沒有具體實現,直接拋了異常,需要子類重寫
    -wait:必須有的,返回一個數位,告訴前端,還有多長時間能存取
    -------都是為了給allow_request輔助的-----
    get_rate
    parse_rate
    throttle_failure
    throttle_success
    """

def allow_request(self, request, view):
        if self.rate is None: # 要取init中看,self.rate=3/m,如果自己在頻率類中寫rate=5/m,我們就不需要寫scope和組態檔了
            return True
        # get_cache_key返回了ip:192.168.1.19
        self.key = self.get_cache_key(request, view)
        if self.key is None: # 如果get_cache_key返回None,也不會做頻率限制
            return True
        # self.history = self.VISIT_RECORD.get(self.key, [])
        # self.cache 是快取,相當於咱們的self.VISIT_RECORD
        self.history = self.cache.get(self.key, [])
        self.now = self.timer()# 取出當前時間
        # self.duration:設定的1分鐘存取3次,self.duration就是60
        while self.history and self.history[-1] <= self.now - self.duration:
            self.history.pop()
        if len(self.history) >= self.num_requests: # '5/m',num_requests就是5
            return self.throttle_failure()
        return self.throttle_success()

# __init__    
def __init__(self):
  if not getattr(self, 'rate', None): # 如果沒有,執行下面的程式碼
      self.rate = self.get_rate()
  self.num_requests, self.duration = self.parse_rate(self.rate)

# self.get_rate()
    def get_rate(self):
        # self.scope  我們自己寫的字串
        # self.THROTTLE_RATES組態檔中的那個字典
        return self.THROTTLE_RATES[self.scope]  # 咱們設定的 3/m

# self.num_requests, self.duration = self.parse_rate(self.rate)---》self.rate是3/s
    def parse_rate(self, rate):
        if rate is None:
            return (None, None)
        # 3/m---->num=3,period=m
        num, period = rate.split('/')
        # num_requests=數位3
        num_requests = int(num)
        # 
        d={'s': 1, 'm': 60, 'h': 3600, 'd': 86400}
        period[0]='m'
        d[m]
        duration = {'s': 1, 'm': 60, 'h': 3600, 'd': 86400}[period[0]]
        return (num_requests, duration)