1. 程式人生 > >drf頻率元件

drf頻率元件

一 自定義頻率類,自定義頻率規則

自定義的邏輯

#(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)