Django-Rest frameworw用户访问次数/频率限制

一、示例展示
=============******=============
            1、BaseThrottle原类

            2、基于用户IP限制访问频率 （了解）

            3、基于用户IP显示访问频率（利于Django缓存） # 经常使用

            4、匿名时用IP限制+登录时用Token限制

=============******=============

全局使用：
in settings.pyREST_FRAMEWORK = {
'DEFAULT_THROTTLE_CLASSES': [
'api.utils.throttles.throttles.LuffyAnonRateThrottle',
'api.utils.throttles.throttles.LuffyUserRateThrottle',
],
'DEFAULT_THROTTLE_RATES': {
'anon': '10/day',
'user': '10/day',
'luffy_anon': '10/m',
'luffy_user': '20/m',
},
}

1、BaseThrottle原类class BaseThrottle(object):
"""
Rate throttling of requests.
"""

def allow_request(self, request, view): #判断是否允许请求，来访问
"""
Return `True` if the request should be allowed, `False` otherwise.
"""
raise NotImplementedError('.allow_request() must be overridden')

def get_ident(self, request): # 获取请求的IP，
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): # 等待时间设置
"""
Optionally, return a recommended number of seconds to wait before
the next request.
"""
return None
2、基于用户IP限制访问频率 （了解）in url.py
from django.conf.urls import url, include
from web.views import TestView

urlpatterns = [
url(r'^test/', TestView.as_view()),
]
in views.py
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
from rest_framework.views import APIView
from rest_framework.response import Response

from rest_framework import exceptions
from rest_framework.throttling import BaseThrottle
from rest_framework.settings import api_settings

# 保存访问记录
RECORD = {
'用户IP': [12312139, 12312135, 12312133, ]
}

class TestThrottle(BaseThrottle):
'''仿照SimpleRateThrottle这个类，做简单化处理，看懂后在看SimpleRateThrottle就很简单了'''
ctime = time.time

def get_ident(self, request):
"""
根据用户IP和代理IP，当做请求者的唯一IP
"""
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 allow_request(self, request, view):
"""
是否仍然在允许范围内
:return: True，表示可以通过；False表示已超过限制，不允许访问
"""
# 获取用户唯一标识（如：IP）

# 允许一分钟访问10次
num_request = 10
time_request = 60

now = self.ctime()
ident = self.get_ident(request)
self.ident = ident
if ident not in RECORD:
RECORD[ident] = [now, ]
return True
history = RECORD[ident]
while history and history[-1] <= now - time_request:
history.pop()
if len(history) < num_request:
history.insert(0, now)
return True

def wait(self):
"""
多少秒后可以允许继续访问
"""
last_time = RECORD[self.ident][0]
now = self.ctime()
return int(60 + last_time - now)

class TestView(APIView):
throttle_classes = [TestThrottle, ]

def get(self, request, *args, **kwargs):
# self.dispatch
print(request.user)
print(request.auth)
return Response('GET请求，响应内容')

def post(self, request, *args, **kwargs):
return Response('POST请求，响应内容')

def put(self, request, *args, **kwargs):
return Response('PUT请求，响应内容')

def throttled(self, request, wait):
"""
访问次数被限制时，定制错误信息
"""

class Throttled(exceptions.Throttled):
default_detail = '请求被限制.'
extra_detail_singular = '请 {wait} 秒之后再重试.'
extra_detail_plural = '请 {wait} 秒之后再重试.'

raise Throttled(wait)3、基于用户IP显示访问频率（利于Django缓存） # 经常使用in settings.py
REST_FRAMEWORK = {
'DEFAULT_THROTTLE_RATES': {
'test_scope': '10/m', # s 秒 h 小时 d 天 m 分钟 等等 源码通过/切割，判断右边首位是什么来判断时间单位
},
}

in url.py
from django.conf.urls import url, include
from web.views import TestView

urlpatterns = [
url(r'^test/', TestView.as_view()),
]

in views.py
#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response

from rest_framework import exceptions
from rest_framework.throttling import SimpleRateThrottle

class TestThrottle(SimpleRateThrottle):

scope = 'test_scope' # 显示频率的Key,在配置文件里需要有个跟这个同名

def get_cache_key(self, request, view):
return self.get_ident(request) # 获取请求IP

class TestView(APIView):
throttle_classes = [TestThrottle, ]

def get(self, request, *args, **kwargs):
# self.dispatch
print(request.user)
print(request.auth)
return Response('GET请求，响应内容')

def post(self, request, *args, **kwargs):
return Response('POST请求，响应内容')

def put(self, request, *args, **kwargs):
return Response('PUT请求，响应内容')

def throttled(self, request, wait):
"""
访问次数被限制时，定制错误信息
"""

class Throttled(exceptions.Throttled):
default_detail = '请求被限制.'
extra_detail_singular = '请 {wait} 秒之后再重试.'
extra_detail_plural = '请 {wait} 秒之后再重试.'

raise Throttled(wait)4、匿名时用IP限制+登录时用Token限制

in settings.py
REST_FRAMEWORK = {
'UNAUTHENTICATED_USER': None,
'UNAUTHENTICATED_TOKEN': None,
'DEFAULT_THROTTLE_RATES': {
'luffy_anon': '10/m',
'luffy_user': '20/m',
},
}
in url.py
from django.conf.urls import url, include
from web.views.s3_throttling import TestView

urlpatterns = [
url(r'^test/', TestView.as_view()),
]
in views.py
#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response

from rest_framework.throttling import SimpleRateThrottle

class LuffyAnonRateThrottle(SimpleRateThrottle):
"""
匿名用户，根据IP进行限制
"""
scope = "luffy_anon"

def get_cache_key(self, request, view):
# 用户已登录，则跳过 匿名频率限制
if request.user:
return None

return self.cache_format % {
'scope': self.scope,
'ident': self.get_ident(request)
}

class LuffyUserRateThrottle(SimpleRateThrottle):
"""
登录用户，根据用户token限制
"""
scope = "luffy_user"

def get_ident(self, request):
"""
认证成功时：request.user是用户对象；request.auth是token对象
:param request:
:return:
"""
# return request.auth.token
return "user_token"

def get_cache_key(self, request, view):
"""
获取缓存key
:param request:
:param view:
:return:
"""
# 未登录用户，则跳过 Token限制
if not request.user:
return None

return self.cache_format % {
'scope': self.scope,
'ident': self.get_ident(request)
}

class TestView(APIView):
throttle_classes = [LuffyUserRateThrottle, LuffyAnonRateThrottle, ]

def get(self, request, *args, **kwargs):
# self.dispatch
print(request.user)
print(request.auth)
return Response('GET请求，响应内容')

def post(self, request, *args, **kwargs):
return Response('POST请求，响应内容')

def put(self, request, *args, **kwargs):
return Response('PUT请求，响应内容')