1、节流源码流程:
# step 1
def dispatch(self, request, *args, **kwargs):
try:
self.initial(request, *args, **kwargs)
# step 2
def initial(self, request, *args, **kwargs):
self.perform_authentication(request) # 先认证
self.check_permissions(request) # 随后权限管理
self.check_throttles(request) # 最后是节流管理
# step 3
def check_throttles(self, request):
for throttle in self.get_throttles():
# 调用类的allow_request函数,返回False,则节流有效,反之,不节流
if not throttle.allow_request(request, self):
self.throttled(request, throttle.wait())
# step 4 :获取节流类的实例化对象列表
def get_throttles(self):
return [throttle() for throttle in self.throttle_classes]
# step 5
from rest_framework.throttling import BaseThrottle
# 默认的节流类
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 get_cache_key(self, request, view):
pass
2、自定义节流:
# 相关参数的解析方式
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)
# setting.py
REST_FRAMEWORK = {
'DEFAULT_THROTTLE_RATES':{
'anon':'5/m' # 表示每分钟只能访问5次数
},
}
# views.py
class MyThrottle(SimpleRateThrottle):
scope = 'anon' # 标识符
def get_cache_key(self, request, view):
return self.cache_format % {
'scope': self.scope,
'ident': self.get_ident(request)
}
class Throttle(object):
throttle_classes = [MyThrottle,]
class UserView(Auth,Throttle,APIView):
"""所有用户都有权限,也可以限制匿名用户,去掉Auth类参数即可"""
def get(self,request,*args,**kwargs):
return HttpResponse('<h1>用户界面</h1>')
3、错误页面:
4、请求次数限制方法,可以对ip(对代理用户无效)、用户名或者手机等特定标识节流:
dict={
ip:[11:12,11:10,11:01,] # 最新访问时间从左侧插入
# 例如若限制为:5/m,那么每次插入元素前都会与相应的元素比较时间差,若在时间差内则插入,否则不插入
}
原文地址:https://www.cnblogs.com/fqh202/p/9398364.html
时间: 2024-10-22 04:08:09