标签:需要 lse time none while ttl 实现 his 使用
列表从后往前读
#1.在request中获取IP
#2.访问记录
VISIT_RECORD = {} 放缓存 数据库 都可以 建议缓存
import time
class VisitThrottle(object):
"""10s内只能访问3次"""
def __init__(self):
self.history = None
def allow_request(self, request, view):
#1.在request中获取IP
#2.访问记录
#remote_addr = request._request.META.get(‘REMOTE_ADDR‘)
remote_addr = request.META.get("REMOTE_ADDR") #都可以
ctime = time.time()
if remote_addr not in VISIT_RECORD:
VISIT_RECORD[remote_addr] = [ctime,]
history = VISIT_RECORD.get(remote_addr)
self.history = history
while history and history[-1] < ctime - 10:
history.pop()
if len(history) < 3:
history.insert(0, ctime)
return True
return True #表示可以继续访问
#return False #访问频率太高,被限制
def wait(self):
"""还需要等多少秒"""
ctime = time.time()
data = 60 - (ctime - self.history[-1])
return data
throttle_classes = [VisitThrottle,]
#全局配置
REST_FRAMEWORK = {
"DEFAULT_THROTTLE_CLASSES":[‘api.utils.throttle.VisitThrottle‘]
}
源码流程
check_throttles
self.get_throttles
内置函数
from rest_framework.throttling import BaseThrottle
class BaseThrottle(object):
def allow_request(self,request, view) #由这个函数进行触发
def get_ident(self,request):
def wait(self):
到时可以使用SimpleRateThrottle
from rest_framework.throttling import SimpleRateThrottle
class VisiThrottle(SimpleRateThrottle):
scope = "xiao" #当KEY使用 当在setting设置DEFAULT_THROTTLE_RATES : {"xiao":‘3/m‘} m分 h时 d天
def get_cache_key(self, request, view):
return request.META.get("REMOTE_ADDR")
class VIPThrottle(SimpleRateThrottle):
scope = "vip" #当KEY使用 当在setting设置DEFAULT_THROTTLE_RATES : {"xiao":‘3/m‘} m分 h时 d天
def get_cache_key(self, request, view):
return request.user.username #认证时的对象的username
Settings里面添加
REST_FRAMEWORK = {
"DEFAULT_THROTTLE_CLASSES":[‘cmdb.utils.throttle.Visit2Throttle‘], #此时不能在这里添加两个控制
"DEFAULT_THROTTLE_RATES": {
"xiao": ‘3/m‘,
"vip":‘10/m‘, #vip用户访问频率限制
}
}
基本使用
类 继承:BaseThrottle 实现:allow_request, wait
类 继承:SimpleRateThrottle 实现:get_cache_key 、 scope = "xiao" (配置文件中的key)
标签:需要 lse time none while ttl 实现 his 使用
原文地址:https://www.cnblogs.com/Liang-jc/p/9385465.html