|
目錄
頻率組件
一、作用
為了控制用戶(hù)對(duì)某個(gè)url請(qǐng)求的頻率，比如，一分鐘以?xún)?nèi)，只能訪問(wèn)三次
二、自定義頻率類(lèi)
# 寫(xiě)一個(gè)頻率認(rèn)證類(lèi)
class MyThrottle:
visit_dic = {}
visit_time = None
def __init__(self):
self.ctime = time.time()
# 重寫(xiě)allow_request()方法
# request是request對(duì)象，view是視圖類(lèi)，可以對(duì)視圖類(lèi)進(jìn)行操作
def allow_request(self, request, view):
'''
（1）取出訪問(wèn)者ip
（2）判斷當(dāng)前ip不在訪問(wèn)字典里，添加進(jìn)去，并且直接返回True,表示第一次訪問(wèn)，在字典里，繼續(xù)往下走
（3）循環(huán)判斷當(dāng)前ip的列表，有值，并且當(dāng)前時(shí)間減去列表的最后一個(gè)時(shí)間大于60s，把這種數(shù)據(jù)pop掉，這樣列表中只有60s以?xún)?nèi)的訪問(wèn)時(shí)間，
（4）判斷，當(dāng)列表小于3，說(shuō)明一分鐘以?xún)?nèi)訪問(wèn)不足三次，把當(dāng)前時(shí)間插入到列表第一個(gè)位置，返回True，順利通過(guò)
（5）當(dāng)大于等于3，說(shuō)明一分鐘內(nèi)訪問(wèn)超過(guò)三次，返回False驗(yàn)證失敗
visit_dic = {ip1:[time2, time1, time0],
ip2:[time1, time0],
}
'''
# 取出訪問(wèn)者ip，ip可以從請(qǐng)求頭中取出來(lái)
ip = request.META.get('REMOTE_ADDR')
# 判斷該次請(qǐng)求的ip是否在地點(diǎn)中
if ip in self.visit_dic:
# 當(dāng)存在字典中時(shí)，取出這個(gè)ip訪問(wèn)時(shí)間的列表
visit_time = self.visit_dic[ip]
self.visit_time = visit_time
while visit_time:
# 當(dāng)訪問(wèn)時(shí)間列表中有值，時(shí)間間隔超過(guò)60，就將那個(gè)歷史時(shí)間刪除
if self.ctime - visit_time[-1] > 60:
visit_time.pop()
else:
# 當(dāng)pop到一定時(shí)，時(shí)間間隔不大于60了，退出循環(huán)，此時(shí)得到的是60s內(nèi)訪問(wèn)的時(shí)間記錄
break
# while循環(huán)等價(jià)于
# while visit_time and ctime - visit_time[-1] > 60:
# visit_time.pop()
# 列表長(zhǎng)度可表示訪問(wèn)次數(shù)，根據(jù)源碼，可以得出，返回值是Boolean類(lèi)型
if len(visit_time) >= 3:
return False
else:
# 如果60秒內(nèi)訪問(wèn)次數(shù)小于3次，將當(dāng)前訪問(wèn)的時(shí)間記錄下來(lái)
visit_time.insert(0, self.ctime)
return True
else:
# 如果字典中沒(méi)有當(dāng)前訪問(wèn)ip，將ip加到字典中
self.visit_dic[ip] = [self.ctime, ]
return True
# 獲取下次距訪問(wèn)的時(shí)間
def wait(self):
return 60 - (self.ctime - self.visit_time[-1])
# view層
from app01 import MyAuth
from rest_framework import exceptions
class Book(APIView):
# 局部使用頻率控制
throttle_classes = [MyAuth.MyThrottle, ]
def get(self,request):
return HttpResponse('ok')
# 重寫(xiě)拋出異常的方法 throttled
def throttled(self, request, wait):
class MyThrottled(exceptions.Throttled):
default_detail = '下次訪問(wèn)'
extra_detail_singular = '還剩 {wait} 秒.'
extra_detail_plural = '還剩 {wait} 秒'
raise MyThrottled(wait)
三、內(nèi)置的訪問(wèn)頻率控制類(lèi)
from rest_framework.throttling import SimpleRateThrottle
# 寫(xiě)一個(gè)頻率控制類(lèi)，繼承SimpleRateThrottle類(lèi)
class MyThrottle(SimpleRateThrottle):
# 配置scope，通過(guò)scope到setting中找到 3/m
scope = 'ttt'
def get_cache_key(self, request, view):
# 返回ip，效果和 get_ident() 方法相似
# ip = request.META.get('REMOTE_ADDR')
# return ip
# get_ident 返回的就是ip地址
return self.get_ident(request)
# view層視圖類(lèi)
class Book(APIView):
throttle_classes = [MyAuth.MyThrottle, ]
def get(self, request):
return HttpResponse('ok')
def throttled(self, request, wait):
class MyThrottled(exceptions.Throttled):
default_detail = '下次訪問(wèn)'
extra_detail_singular = '還剩 {wait} 秒.'
extra_detail_plural = '還剩 {wait} 秒'
raise MyThrottled(wait)
# setting中配置
REST_FRAMEWORK = {
'DEFAULT_THROTTLE_RATES': {
'ttt': '10/m'
}
}
四、全局、局部使用
1、全局使用
在setting中配置
REST_FRAMEWORK = {
'DEFAULT_THROTTLE_CLASSES': ['app01.MyAuth.MyThrottle', ],
}
2、局部使用
在視圖類(lèi)中重定義throttle_classes
throttle_classes = [MyAuth.MyThrottle, ]
3、局部禁用
在視圖類(lèi)中重定義throttle_classes為一個(gè)空列表
throttle_classes = []
五、源碼分析
1、as_view -----> view ------> dispatch ------> initial ----> check_throttles 頻率控制
2、self.check_throttles(request)
def check_throttles(self, request):
"""
Check if request should be throttled.
Raises an appropriate exception if the request is throttled.
"""
# （2-----1） get_throttles 由頻率類(lèi)產(chǎn)生的對(duì)象組成的列表
for throttle in self.get_throttles():
if not throttle.allow_request(request, self):
# （4）異常信息的處理
self.throttled(request, throttle.wait())
（2-----1） self.get_throttles()
def get_throttles(self):
"""
Instantiates and returns the list of throttles that this view uses.
"""
return [throttle() for throttle in self.throttle_classes]
3、allow_request()
自身、所在類(lèi)找都沒(méi)有，去父類(lèi)中找
class SimpleRateThrottle(BaseThrottle):
cache = default_cache
timer = time.time
cache_format = 'throttle_%(scope)s_%(ident)s'
scope = None
THROTTLE_RATES = api_settings.DEFAULT_THROTTLE_RATES
def __init__(self):
if not getattr(self, 'rate', None):
self.rate = self.get_rate()
self.num_requests, self.duration = self.parse_rate(self.rate)
def parse_rate(self, rate):
if rate is None:
return (None, None)
num, period = rate.split('/')
num_requests = int(num)
duration = {'s': 1, 'm': 60, 'h': 3600, 'd': 86400}[period[0]]
return (num_requests, duration)
def allow_request(self, request, view):
if self.rate is None:
return True
# （3-----1） get_cache_key就是要重寫(xiě)的方法，若不重寫(xiě)，會(huì)直接拋出異常
self.key = self.get_cache_key(request, view)
if self.key is None:
return True
self.history = self.cache.get(self.key, [])
self.now = self.timer()
# Drop any requests from the history which have now passed the
# throttle duration
while self.history and self.history[-1] <= self.now - self.duration:
self.history.pop()
if len(self.history) >= self.num_requests:
return self.throttle_failure()
return self.throttle_success()
# 返回距下一次能請(qǐng)求的時(shí)間
def wait(self):
"""
Returns the recommended next request time in seconds.
"""
if self.history:
remaining_duration = self.duration - (self.now - self.history[-1])
else:
remaining_duration = self.duration
（3-----1） self.get_cache_key(request, view)
def get_cache_key(self, request, view):
"""
Should return a unique cache-key which can be used for throttling.
Must be overridden.
May return `None` if the request should not be throttled.
"""
raise NotImplementedError('.get_cache_key() must be overridden')
4、self.throttled(request, throttle.wait()) --------> 拋出異常
def throttled(self, request, wait):
"""
If request is throttled, determine what kind of exception to raise.
"""
raise exceptions.Throttled(wait)
（4------1）raise exceptions.Throttled(wait) -------> 異常信息
class Throttled(APIException):
status_code = status.HTTP_429_TOO_MANY_REQUESTS
# 重寫(xiě)下面三個(gè)變量就可以修改顯示的異常信息，例如用中文顯示異常信息
default_detail = _('Request was throttled.')
extra_detail_singular = 'Expected available in {wait} second.'
extra_detail_plural = 'Expected available in {wait} seconds.'
default_code = 'throttled'
def __init__(self, wait=None, detail=None, code=None):
if detail is None:
detail = force_text(self.default_detail)
if wait is not None:
wait = math.ceil(wait)
detail = ' '.join((
detail,
force_text(ungettext(self.extra_detail_singular.format(wait=wait),
self.extra_detail_plural.format(wait=wait),
wait))))
self.wait = wait
super(Throttled, self).__init__(detail, code)
來(lái)源：http://www./content-4-92401.html
|
