drf頻率元件
阿新 • • 發佈:2018-12-14
一 自定義頻率類,自定義頻率規則
自定義的邏輯
#(1)取出訪問者ip # (2)判斷當前ip不在訪問字典裡,新增進去,並且直接返回True,表示第一次訪問,在字典裡,繼續往下走 # (3)迴圈判斷當前ip的列表,有值,並且當前時間減去列表的最後一個時間大於60s,把這種資料pop掉,這樣列表中只有60s以內的訪問時間, # (4)判斷,當列表小於3,說明一分鐘以內訪問不足三次,把當前時間插入到列表第一個位置,返回True,順利通過 # (5)當大於等於3,說明一分鐘內訪問超過三次,返回False驗證失敗
程式碼實現
import time class MyThrottles(): visitor_dic= {} def __init__(self): self.history=None def allow_request(self,request, view): #(1)取出訪問者ip # print(request.META) ip=request.META.get('REMOTE_ADDR') ctime=time.time() # (2)判斷當前ip不在訪問字典裡,新增進去,並且直接返回True,表示第一次訪問 if ip not in self.visitor_dic: self.visitor_dic[ip]=[ctime,] return True history = self.visitor_dic[ip] self.history=history # (3)迴圈判斷當前ip的列表,有值,並且當前時間減去列表的最後一個時間大於60s,把這種資料pop掉,這樣列表中只有60s以內的訪問時間, while history and ctime-history[-1]>60: history.pop() # (4)判斷,當列表小於3,說明一分鐘以內訪問不足三次,把當前時間插入到列表第一個位置,返回True,順利通過# (5)當大於等於3,說明一分鐘內訪問超過三次,返回False驗證失敗 if len(history)<3: history.insert(0,ctime) return True else: return False def wait(self): ctime=time.time() return 60-(ctime-self.history[-1])
views:
class Test(APIView):
throttle_classes = [MyThrottle, ]
def get(self, request):
return HttpResponse('ok')
二 內建頻率類
#內建的訪問頻率控制類SimpleRateThrottle #寫一個類, 繼承SimpleRateThrottle from rest_framework.throttling import BaseThrottle,SimpleRateThrottle class MyThrottle(SimpleRateThrottle): scope = 'aaa' def get_cache_key(self, request, view): # 返回ip地址,兩種方法: # ip=request.META.get('REMOTE_ADDR') # return ip return self.get_ident(request) #在setting中: REST_FRAMEWORK = { 'DEFAULT_THROTTLE_RATES': { 'aaa': '10/m' } } #區域性使用,在檢視類中寫 throttle_classes = [MyThrottle, ] #全域性使用,在setting中寫 REST_FRAMEWORK = { 'DEFAULT_THROTTLE_CLASSES': ['app01.MyAuth.MyThrottle', ], } #區域性禁用,在檢視類中寫 throttle_classes = []
from django.shortcuts import render, HttpResponse # Create your views here. from rest_framework.views import APIView from rest_framework import exceptions from app01.MyAuth import MyThrottle from rest_framework.request import Request from django.core.handlers.wsgi import WSGIRequest from rest_framework.parsers import JSONParser,FormParser class Test(APIView): # throttle_classes = [MyThrottle,] # parser_classes = [FormParser,] def get(self, request): print(type(request._request)) return HttpResponse('ok') def post(self,request): print(request.data) return HttpResponse('post') #等待時間顯示中文資訊 def throttled(self, request, wait): class MyThrottled(exceptions.Throttled): default_detail = '傻子' extra_detail_singular = '還剩 {wait} 秒才可以訪問.' extra_detail_plural = '還剩 {wait} 秒才可以訪問' raise MyThrottled(wait)