python_restframework
2023-01-31 01:01:31
python
(1) 取出访问者ip
(2) 判断当前ip不在访问字典里，添加进去，并且直接返回True,表示第一次访问，在字典里，继续往下走
(3) 循环判断当前ip的列表，有值，并且当前时间减去列表的最后一个时间大于60s，把这种数据pop掉，这样列表中只有60s以内的访问时间，
(4) 判断，当列表小于3，说明一分钟以内访问不足三次，把当前时间插入到列表第一个位置，返回True，顺利通过
(5) 当大于等于3，说明一分钟内访问超过三次，返回False验证失败1、分发display
def dispatch(self, request, *args, **kwargs):
try:
# 进入初始化
self.initial(request, *args, **kwargs)2、 drf初始化方法
apiview下的方法
def initial(self, request, *args, **kwargs):
# 认证
self.perfORM_authentication(request)
# 进入权限
self.check_permissions(request)
# --> 频率
self.check_throttles(request)3、频率模块
def check_throttles(self, request):
"""
Check if request should be throttled.
Raises an appropriate exception if the request is throttled.
"""
# 循环的是列表
for throttle in self.get_throttles():
# 返回结果true or false， false就继续执行
if not throttle.allow_request(request, self):
# 视图类的三个参数， self, request，
# throttle.wait()： 最后wait返回的数字
self.throttled(request, throttle.wait())3.1、for throttle in self.get_throttles():
def get_throttles(self):
"""
Instantiates and returns the list of throttles that this view uses.
"""
# 跟权限组件一样， 这里循环 return出去的也是一个列表
return [throttle() for throttle in self.throttle_classes]3.2、if 判断
# 当返回为false时，说明该用户或订单无此权限,
# not false为true 继续往下执行, not true就是 false, 不执行下面代码
if not throttle.allow_request(request, self):3.3、throttle.allow_request
# 如果是false 就直接返回这个错误了
def allow_request(self, request, view):
"""
Return `True` if the request should be allowed, `False` otherwise.
"""
raise NotImplementedError('.allow_request() must be overridden')
4、BasePermission
# 继承基础的认证权限， 如果忘了要定义哪个类 直接在这里看也OK
from rest_framework.throttling import BaseThrottle
# 自定义的组件
def allow_request(self, request, view):
raise NotImplementedError('.allow_request() must be overridden')
def get_ident(self, request):
xff = request.META.get('Http_X_FORWARDED_FOR')
remote_addr = request.META.get('REMOTE_ADDR')
num_proxies = api_settings.NUM_PROXIES
if num_proxies is not None:
if num_proxies == 0 or xff is None:
return remote_addr
addrs = xff.split(',')
client_addr = addrs[-min(num_proxies, len(addrs))]
return client_addr.strip()
return ''.join(xff.split()) if xff else remote_addr
# 最后要返回的时间或者其它
def wait(self):
return None
5、定义一个权限
class MyPermission(BasePermission):
# 前台返回的错误
message = "您没有权限, 请先联系下管理员增加权限"
# 获取权限
def has_permission(self,request, view):
# 认证组件， 返回的request.user
print("permission: ", request.user.permission)
if request.user.permission > 1:
# 如果大于就是true， 在第3步，if not true 等于false就不执行它了
return True
return False6、频率组件
class FirstThrottle(BaseThrottle):
get_ip = {}
def __init__(self):
self.history = None
self.ip = None
self.ctime = time.time()
def allow_request(self, request, view):
:param request: 浏览器请求过来的数据
:param view: apiview视图
:return: true or false
# 1、取出访问者的IP
client_ip = request.META.get("REMOTE_ADDR")
self.ip = client_ip
# 2、判断不存在的话添加到字典 并将时间也一并添加进去
if client_ip not in self.get_ip:
self.get_ip[client_ip] = [self.ctime, ]
return True
# 获取当前IP的访问时间记录
self.history = self.get_ip[client_ip]
# 3、 开始循环判断, 如果最后一个大于60秒就直接干掉
while self.history and self.ctime - self.history[-1] > 60:
self.history.pop()
if len(self.history) < 3:
self.history.insert(0, self.ctime)
return True
return False
def wait(self):
last_time = self.ctime - self.history[-1] - 10
if last_time == 0:
self.get_ip[self.ip].clear()
return last_time7、全局使用频率
# settings.py文件中定义， 所有的组件都可以放在这里
REST_FRAMEWORK = {
"DEFAULT_THROTTLE_CLASSES": [
'app01.myauth.FirstThrottle', # 全局使用权限
]
}7、局部使用
类中直接使用
throttle_classes = [FirstThrottle, ]8、局部禁用
类中直接使用
throttle_classes = []使用组件中自带的频率控制组件
先在settings.py中定义限制频率的范围 REST_FRAMEWORK={ "DEFAULT_THROTTLE_RATES": { "thro_rate": "10/m" } }
1、进入频率
class SimpleRateThrottle(BaseThrottle):
cache = default_cache
# 获取时间
timer = time.time
cache_format = 'throttle_%(scope)s_%(ident)s'
# 这个是在setting中设置的 DEFAULT_THROTTLE_RATES的字典key， 必须得定义
scope = None
THROTTLE_RATES = api_settings.DEFAULT_THROTTLE_RATES
# 初始化，
def __init__(self):
# 首先就是先判断 rate是否为空, 如果是false为空，就进入self.get_rate
if not getattr(self, 'rate', None):
# 直接输出错误
self.rate = self.get_rate()
# 如果上一步通过，就继续进入这里 9.2
self.num_requests, self.duration = self.parse_rate(self.rate)
# 也就是说执行完9.2之后 获取到的结果就是
# self.num_requests, self.duration = (10,60)1.1、get_rate
def get_rate(self):
# scope 这个值在类中必须被定义成 settings中定义的值 如thro_rate
if not getattr(self, 'scope', None):
msg = ("You must set either `.scope` or `.rate` for '%s' throttle" %
self.__class__.__name__)
raise ImproperlyConfigured(msg)
try:
# 在配置文件中 将thro_rate 取出， 返回 10/m
return self.THROTTLE_RATES[self.scope]
except KeyError:
msg = "No default throttle rate set for '%s' scope" % self.scope
raise ImproperlyConfigured(msg)
2、当初始化通过
self.num_requests, self.duration = self.parse_rate(self.rate)
def parse_rate(self, rate):
"""
Given the request rate string, return a two tuple of:
<allowed number of requests>, <period of time in seconds>
"""
# 这个是在setting中设置的 DEFAULT_THROTTLE_RATES的字典设置为空，就直接返回none,none
if rate is None:
return (None, None)
# 这里的rate就是就是get_rate取出来的10/m 然后切割它
num, period = rate.split('/')
num_requests = int(num)
# 定义如果是m就是60秒，然后字典中直接取值这里是m取出来的就是60
duration = {'s': 1, 'm': 60, 'h': 3600, 'd': 86400}[period[0]]
# 最后返回它俩
return (num_requests, duration)3、类中调用get_cache_key
def get_cache_key(self, request, view):
"""
# 应返回可用于限制的唯一缓存键。
Should return a unique cache-key which can be used for throttling.
# 必须要重写， 否则调用SimpleRateThrottle也会直接报错
Must be overridden.
May return `None` if the request should not be throttled.
"""
raise NotImplementedError('.get_cache_key() must be overridden')4、实例
class FirstThrottle(SimpleRateThrottle):
# 这里会调用 self.get_rate那个函数，返回的就是 10/m了
scope = "thro_rate"
# 如果不重新定义就会报错， 因为它得从缓存中找出 ip地址
def get_cache_key(self, request, view):
# 返回空也行， 也会有倒计时
return self.get_ident(request)
# "detail": "Request was throttled. Expected available in 56 seconds."5、中文显示错误日志
5.1、流程的前3步
def check_throttles(self, request):
"""
Check if request should be throttled.
Raises an appropriate exception if the request is throttled.
"""
for throttle in self.get_throttles():
if not throttle.allow_request(request, self):
# 如果不存在 就进入到 throttled中
self.throttled(request, throttle.wait())5.2、throttled 错误提示
def throttled(self, request, wait):
"""
If request is throttled, determine what kind of exception to raise.
"""
# 返回错误信息
raise exceptions.Throttled(wait)5.3、重写exceptions方法
class Throttled(APIException):
status_code = status.HTTP_429_TOO_MANY_REQUESTS
default_detail = _('Request was throttled.')
extra_detail_singular = 'Expected available in {wait} second.'
extra_detail_plural = 'Expected available in {wait} seconds.'5.4、实例
from app01.SelfThrottle import FirstThrottle
from rest_framework import exceptions
class Thor(APIView):
# 局部使用
throttle_classes = [FirstThrottle, ]
def get(self, request, *args, **kwargs):
return HttpResponse("ok")
# 需要注意的是 这里需要在视图类的重写方法，或继承
def throttled(self, request, wait):
class Myerrors(exceptions.Throttled):
default_detail = "超出频率限制"
extra_detail_singular = '请 {wait} 秒后在访问.'
extra_detail_plural = '请 {wait} 秒后在访问.'
raise Myerrors(wait)相关文章