一、认证和授权

a. 用户url传入的token认证

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

rest_framework.views rest_framework.response rest_framework.authentication rest_framework.permissions from rest_framework.request import Request
from rest_framework import exceptions

token_list = [
'sfsfss123kuf3j123','asijnfowerkkf9812',]

class TestAuthentication(BaseAuthentication):
def authenticate(self,request):
"""
用户认证，如果验证成功后返回元组： (用户,用户Token)
:param request:
:return:
None,表示跳过该验证；
如果跳过了所有认证，默认用户和Token和使用配置文件进行设置
self._authenticator = None
if api_settings.UNAUTHENTICATED_USER:
self.user = api_settings.UNAUTHENTICATED_USER() # 默认值为：匿名用户
else:
self.user = None

            if api_settings.UNAUTHENTICATED_TOKEN:
                self.auth = api_settings.UNAUTHENTICATED_TOKEN()# <a href="/tag/mo/" target="_blank" class="keywords">默</a>认值为：None
            else:
                self.auth = None
        (user,token)表示验证通过并设置<a href="/tag/yonghuming/" target="_blank" class="keywords">用户名</a>和Token；
        Authentication<a href="/tag/Failed/" target="_blank" class="keywords">Failed</a>异常
    </span><span style="color: #800000;"&gt;"""</span><span style="color: #000000;"&gt;
    val </span>= request.query_p<a href="/tag/ara/" target="_blank" class="keywords">ara</a>ms.get(<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;token</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;)
    </span><span style="color: #0000ff;"&gt;if</span> val <span style="color: #0000ff;"&gt;not</span> <span style="color: #0000ff;"&gt;in</span><span style="color: #000000;"&gt; token_list:
        </span><span style="color: #0000ff;"&gt;raise</span> exceptions.Authentication<a href="/tag/Failed/" target="_blank" class="keywords">Failed</a>(<span style="color: #800000;"&gt;"</span><span style="color: #800000;"&gt;<a href="/tag/yonghu/" target="_blank" class="keywords">用户</a>认证失败</span><span style="color: #800000;"&gt;"</span><span style="color: #000000;"&gt;)

    </span><span style="color: #0000ff;"&gt;return</span> (<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;<a href="/tag/denglu/" target="_blank" class="keywords">登录</a><a href="/tag/yonghu/" target="_blank" class="keywords">用户</a></span><span style="color: #800000;"&gt;'</span>,<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;<a href="/tag/yonghu/" target="_blank" class="keywords">用户</a>token</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;)

</span><span style="color: #0000ff;"&gt;def</span><span style="color: #000000;"&gt; authenticate_header(self,request):
    </span><span style="color: #800000;"&gt;"""</span><span style="color: #800000;"&gt;
    Return a string to be used as the value of the `WWW-Authenticate`
    header in a `401 Unauthenticated` response,or `None` if the
    authentication scheme should return `403 Permission Denied` responses.
    </span><span style="color: #800000;"&gt;"""</span>
     <span style="color: #008000;"&gt;#</span><span style="color: #008000;"&gt; 验证失败时，返回的响应头WWW-Authenticate对应的值</span>
    <span style="color: #0000ff;"&gt;pass</span>

</span><span style="color: #008000;"&gt;#</span><span style="color: #008000;"&gt; 权限</span>
<span style="color: #008000;"&gt;#</span><span style="color: #008000;"&gt; 循环执行所有的权限</span>
permission_classes =<span style="color: #000000;"&gt; [TestPermission,]

</span><span style="color: #0000ff;"&gt;def</span> get(self,request,*args,**<span style="color: #000000;"&gt;<a href="/tag/kwargs/" target="_blank" class="keywords">kwargs</a>):
    </span><span style="color: #008000;"&gt;#</span><span style="color: #008000;"&gt; self.<a href="/tag/dis/" target="_blank" class="keywords">dis</a>patch</span>
    <span style="color: #0000ff;"&gt;print</span><span style="color: #000000;"&gt;(request.user)
    </span><span style="color: #0000ff;"&gt;print</span><span style="color: #000000;"&gt;(request.auth)
    </span><span style="color: #0000ff;"&gt;return</span> Response(<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;GET请求，响应<a href="/tag/neirong/" target="_blank" class="keywords">内容</a></span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;)

</span><span style="color: #0000ff;"&gt;def</span> post(self,**<span style="color: #000000;"&gt;<a href="/tag/kwargs/" target="_blank" class="keywords">kwargs</a>):
    </span><span style="color: #0000ff;"&gt;return</span> Response(<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;POST请求，响应<a href="/tag/neirong/" target="_blank" class="keywords">内容</a></span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;)

</span><span style="color: #0000ff;"&gt;def</span> put(self,**<span style="color: #000000;"&gt;<a href="/tag/kwargs/" target="_blank" class="keywords">kwargs</a>):
    </span><span style="color: #0000ff;"&gt;return</span> Response(<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;PUT请求，响应<a href="/tag/neirong/" target="_blank" class="keywords">内容</a></span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;)

views.py

b. 请求头认证

django.conf.urls

rest_framework.views rest_framework.response rest_framework.authentication rest_framework.request rest_framework token_list = [
'sfsfss123kuf3j123',表示跳过该验证；
如果跳过了所有认证，默认用户和Token和使用配置文件进行设置
self._authenticator = None
if api_settings.UNAUTHENTICATED_USER:
self.user = api_settings.UNAUTHENTICATED_USER()
else:
self.user = None

            if api_settings.UNAUTHENTICATED_TOKEN:
                self.auth = api_settings.UNAUTHENTICATED_TOKEN()
            else:
                self.auth = None
        (user,token)表示验证通过并设置<a href="/tag/yonghuming/" target="_blank" class="keywords">用户名</a>和Token；
        Authentication<a href="/tag/Failed/" target="_blank" class="keywords">Failed</a>异常
    </span><span style="color: #800000;"&gt;"""</span>
    <span style="color: #0000ff;"&gt;import</span><span style="color: #000000;"&gt; base64
    auth </span>= request.<a href="/tag/Meta/" target="_blank" class="keywords">Meta</a>.get(<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;HTTP_AUTHORIZATION</span><span style="color: #800000;"&gt;'</span>,b<span style="color: #800000;"&gt;''</span><span style="color: #000000;"&gt;)
    </span><span style="color: #0000ff;"&gt;if</span><span style="color: #000000;"&gt; auth:
        auth </span>= auth.encode(<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;utf-8</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;)
    auth </span>=<span style="color: #000000;"&gt; auth.split()
    </span><span style="color: #0000ff;"&gt;if</span> <span style="color: #0000ff;"&gt;not</span> auth <span style="color: #0000ff;"&gt;or</span> auth[0].lower() != b<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;basic</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;:
        </span><span style="color: #0000ff;"&gt;raise</span> exceptions.Authentication<a href="/tag/Failed/" target="_blank" class="keywords">Failed</a>(<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;验证失败</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;)
    </span><span style="color: #0000ff;"&gt;if</span> len(auth) != 2<span style="color: #000000;"&gt;:
        </span><span style="color: #0000ff;"&gt;raise</span> exceptions.Authentication<a href="/tag/Failed/" target="_blank" class="keywords">Failed</a>(<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;验证失败</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;)
    username,part,password </span>= base64.b64decode(auth[1]).decode(<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;utf-8</span><span style="color: #800000;"&gt;'</span>).partition(<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;:</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;)
    </span><span style="color: #0000ff;"&gt;if</span> username == <span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;alex</span><span style="color: #800000;"&gt;'</span> <span style="color: #0000ff;"&gt;and</span> password == <span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;123</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;:
        </span><span style="color: #0000ff;"&gt;return</span> (<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;<a href="/tag/denglu/" target="_blank" class="keywords">登录</a><a href="/tag/yonghu/" target="_blank" class="keywords">用户</a></span><span style="color: #800000;"&gt;'</span>,<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;<a href="/tag/yonghu/" target="_blank" class="keywords">用户</a>token</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;)
    </span><span style="color: #0000ff;"&gt;else</span><span style="color: #000000;"&gt;:
        </span><span style="color: #0000ff;"&gt;raise</span> exceptions.Authentication<a href="/tag/Failed/" target="_blank" class="keywords">Failed</a>(<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;<a href="/tag/yonghuming/" target="_blank" class="keywords">用户名</a>或<a href="/tag/mimap/" target="_blank" class="keywords">密码</a><a href="/tag/cuowu/" target="_blank" class="keywords">错误</a></span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;)

</span><span style="color: #0000ff;"&gt;def</span><span style="color: #000000;"&gt; authenticate_header(self,or `None` if the
    authentication scheme should return `403 Permission Denied` responses.
    </span><span style="color: #800000;"&gt;"""</span>
    <span style="color: #0000ff;"&gt;return</span> <span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;Basic realm=api</span><span style="color: #800000;"&gt;'</span>

class TestView(APIView):
authentication_classes = [TestAuthentication,]
permission_classes = []

</span><span style="color: #0000ff;"&gt;def</span> get(self,**<span style="color: #000000;"&gt;<a href="/tag/kwargs/" target="_blank" class="keywords">kwargs</a>):
    </span><span style="color: #0000ff;"&gt;print</span><span style="color: #000000;"&gt;(request.user)
    </span><span style="color: #0000ff;"&gt;print</span><span style="color: #000000;"&gt;(request.auth)
    </span><span style="color: #0000ff;"&gt;return</span> Response(<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;GET请求，响应<a href="/tag/neirong/" target="_blank" class="keywords">内容</a></span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;)

</span><span style="color: #0000ff;"&gt;def</span> post(self,**<span style="color: #000000;"&gt;<a href="/tag/kwargs/" target="_blank" class="keywords">kwargs</a>):
    </span><span style="color: #0000ff;"&gt;return</span> Response(<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;PUT请求，响应<a href="/tag/neirong/" target="_blank" class="keywords">内容</a></span><span style="color: #800000;"&gt;'</span>)</pre>

c.多个认证规则

django.conf.urls web.views.s2_auth urlpatterns = [
url(r'^test/',]

class Test1Authentication(BaseAuthentication):
def authenticate(self,表示跳过该验证；
如果跳过了所有认证，默认用户和Token和使用配置文件进行设置
self._authenticator = None
if api_settings.UNAUTHENTICATED_USER:
self.user = api_settings.UNAUTHENTICATED_USER() # 默认值为：匿名用户
else:
self.user = None

            if api_settings.UNAUTHENTICATED_TOKEN:
                self.auth = api_settings.UNAUTHENTICATED_TOKEN()# <a href="/tag/mo/" target="_blank" class="keywords">默</a>认值为：None
            else:
                self.auth = None
        (user,b<span style="color: #800000;"&gt;''</span><span style="color: #000000;"&gt;)
    </span><span style="color: #0000ff;"&gt;if</span><span style="color: #000000;"&gt; auth:
        auth </span>= auth.encode(<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;utf-8</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;)
    </span><span style="color: #0000ff;"&gt;else</span><span style="color: #000000;"&gt;:
        </span><span style="color: #0000ff;"&gt;return</span><span style="color: #000000;"&gt; None
    </span><span style="color: #0000ff;"&gt;print</span>(auth,<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;xxxx</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;)
    auth </span>=<span style="color: #000000;"&gt; auth.split()
    </span><span style="color: #0000ff;"&gt;if</span> <span style="color: #0000ff;"&gt;not</span> auth <span style="color: #0000ff;"&gt;or</span> auth[0].lower() != b<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;basic</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;:
        </span><span style="color: #0000ff;"&gt;raise</span> exceptions.Authentication<a href="/tag/Failed/" target="_blank" class="keywords">Failed</a>(<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;验证失败</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;)
    </span><span style="color: #0000ff;"&gt;if</span> len(auth) != 2<span style="color: #000000;"&gt;:
        </span><span style="color: #0000ff;"&gt;raise</span> exceptions.Authentication<a href="/tag/Failed/" target="_blank" class="keywords">Failed</a>(<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;验证失败</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;)
    username,or `None` if the
    authentication scheme should return `403 Permission Denied` responses.
    </span><span style="color: #800000;"&gt;"""</span>
    <span style="color: #008000;"&gt;#</span><span style="color: #008000;"&gt; return 'Basic realm=api'</span>
    <span style="color: #0000ff;"&gt;pass</span>

class Test2Authentication(BaseAuthentication):
def authenticate(self,or None if the
authentication scheme should return 403 Permission Denied responses.
"""
pass

class TestView(APIView):
authentication_classes = [Test1Authentication,Test2Authentication]
permission_classes = []

</span><span style="color: #0000ff;"&gt;def</span> get(self,**<span style="color: #000000;"&gt;<a href="/tag/kwargs/" target="_blank" class="keywords">kwargs</a>):
    </span><span style="color: #0000ff;"&gt;return</span> Response(<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;PUT请求，响应<a href="/tag/neirong/" target="_blank" class="keywords">内容</a></span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;)

views.py

d.认证和权限

django.conf.urls web.views urlpatterns = [
url(r'^test/',]

token_list = [
'sfsfss123kuf3j123',or None if the
authentication scheme should return 403 Permission Denied responses.
"""
pass

class TestPermission(BasePermission):
message = "权限验证失败"

<span style="color: #0000ff;"&gt;def</span><span style="color: #000000;"&gt; has_permission(self,view):
    </span><span style="color: #800000;"&gt;"""</span><span style="color: #800000;"&gt;
    判断是否有权限访问当前请求
    Return `True` if permission is granted,`False` otherwise.
    :p<a href="/tag/ara/" target="_blank" class="keywords">ara</a>m request: 
    :p<a href="/tag/ara/" target="_blank" class="keywords">ara</a>m view: 
    :return: True有权限；False无权限
    </span><span style="color: #800000;"&gt;"""</span>
    <span style="color: #0000ff;"&gt;if</span> request.user == <span style="color: #800000;"&gt;"</span><span style="color: #800000;"&gt;<a href="/tag/guanliyuan/" target="_blank" class="keywords">管理员</a></span><span style="color: #800000;"&gt;"</span><span style="color: #000000;"&gt;:
        </span><span style="color: #0000ff;"&gt;return</span><span style="color: #000000;"&gt; True

</span><span style="color: #008000;"&gt;#</span><span style="color: #008000;"&gt; GenericAPIView中get_object时<a href="/tag/diaoyong/" target="_blank" class="keywords">调用</a></span>
<span style="color: #0000ff;"&gt;def</span><span style="color: #000000;"&gt; has_object_permission(self,view,obj):
    </span><span style="color: #800000;"&gt;"""</span><span style="color: #800000;"&gt;
    视图继承GenericAPIView，并在其中使用get_object时<a href="/tag/huoqu/" target="_blank" class="keywords">获取</a>对象时，触发单独对象权限验证
    Return `True` if permission is granted,`False` otherwise.
    :p<a href="/tag/ara/" target="_blank" class="keywords">ara</a>m request: 
    :p<a href="/tag/ara/" target="_blank" class="keywords">ara</a>m view: 
    :p<a href="/tag/ara/" target="_blank" class="keywords">ara</a>m obj: 
    :return: True有权限；False无权限
    </span><span style="color: #800000;"&gt;"""</span>
    <span style="color: #0000ff;"&gt;if</span> request.user == <span style="color: #800000;"&gt;"</span><span style="color: #800000;"&gt;<a href="/tag/guanliyuan/" target="_blank" class="keywords">管理员</a></span><span style="color: #800000;"&gt;"</span><span style="color: #000000;"&gt;:
        </span><span style="color: #0000ff;"&gt;return</span><span style="color: #000000;"&gt; True

class TestView(APIView):
# 认证的动作是由request.user触发
authentication_classes = [TestAuthentication,**kwargs):
return Response('PUT请求，响应内容')

e.全局使用

上述操作中均是对单独视图进行特殊配置，如果想要对全局进行配置，则需要再配置文件中写入即可。

REST_FRAMEWORK =

django.conf.urls urls.py

rest_framework.views rest_framework.response class TestView(APIView):

</span><span style="color: #0000ff;"&gt;def</span> get(self,**<span style="color: #000000;"&gt;<a href="/tag/kwargs/" target="_blank" class="keywords">kwargs</a>):
    </span><span style="color: #0000ff;"&gt;return</span> Response(<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;PUT请求，响应<a href="/tag/neirong/" target="_blank" class="keywords">内容</a></span><span style="color: #800000;"&gt;'</span>)</pre>

g.自定义认证工作

token = request.query_params.get() arams表示 token == (,) Failed('认证错误') #只要抛出认证错误这样的异常就会去执行下面的函数 APIException(错误 authenticate_header(self,request): authentication_classes = get(self,**kwargs): Response(用户列表)

自定义认证功能

1、需求：Host是匿名用户和用户都能访问 #匿名用户的request.user = none；User只有注册用户能访问

fdbb-46c0-8967-4db0feeeeaff')">

fdbb-46c0-8967-4db0feeeeaff" class="cnblogs_code_hide">

  app03   django.conf.urls  urlpatterns =     
     url(     url(r     url(r     url(r ]

django.shortcuts rest_framework.views APIView rest_framework.response Response rest_framework.authentication BaseAuthentication rest_framework.authentication app01 rest_framework rest_framework.permissions AllowAny 在这个类里面 rest_framework.throttling ottle,SimpleRateThrottle token = request.query_params.get( obj = models.UserInfo.objects.filter(token=irst() obj : 用户名和auth None message = has_permission(self,view): True 匿名用户就说明有权限 False message = has_permission(self,view): request.user== True False authentication_classes = [] 页面不需要认证 dispatch 匿名用户和用户都能访问 #匿名用户的request.user = none 注册用户能访问 authentication_classes = permission_classes = [] Response( 用户能访问,request.user里面有值 authentication_classes = permission_classes = (request.user, Response(用户列表 permission_denied(self,message= mine what kind of exception to raise. request.authenticators exceptions.NotAuthenticated(detail= exceptions.PermissionDenied(detail=message)

用户能访问 message = authentication_classes = [MyAuthentication,] 用户 permission_classes = [MyPermission,AdminPermission,] 用户有没有权限，如果有权限在判断有没有管理员的权限 Response( permission_denied(self,message= mine what kind of exception to raise. request.authenticators exceptions.NotAuthenticated(detail= exceptions.PermissionDenied(detail=message)

如果遇上下面这样的情况，是因为没有通过认证，并且权限中return False了，可以自定制错误信息为中文，参考源码

permission 一个permission对象，调用has_permission =getattr(permission,

permission_denied(self,message=mine what kind of exception to raise. request.authenticators exceptions.PermissionDenied(detail=message)

那么我们可以重写permission_denied这个方法，如下：

用户能访问,request.user里面有值 authentication_classes = permission_classes = Response(用户列表 permission_denied(self,message= mine what kind of exception to raise. request.authenticators exceptions.NotAuthenticated(detail= exceptions.PermissionDenied(detail=message)

2.全局使用

上述操作中均是对单独视图进行特殊配置，如果想要对全局进行配置，则需要再配置文件中写入即可。

REST_FRAMEWORK = : None,匿名用户设置为None , }

authentication_classes = [] 页面不需要认证 dispatch 匿名用户和用户都能访问 #匿名用户的request.user = none 注册用户能访问 authentication_classes = permission_classes = [] Response( 用户能访问,request.user里面有值 authentication_classes = permission_classes = (request.user, Response(用户列表 permission_denied(self,message= mine what kind of exception to raise. request.authenticators exceptions.NotAuthenticated(detail= exceptions.PermissionDenied(detail= 用户能访问 message = authentication_classes = [MyAuthentication,] 用户 permission_classes = [MyPermission,] 用户有没有权限，如果有权限在判断有没有管理员的权限 Response( permission_denied(self,message= mine what kind of exception to raise. request.authenticators exceptions.NotAuthenticated(detail= exceptions.PermissionDenied(detail=message)

用户访问次数/频率限制

1、为什么要限流呢

答：

- 第一点：爬虫，反爬
- 第二点：控制 API 访问次数
- - 登录用户的用户名可以做标识
- 匿名用户可以参考 ip，但是 ip可以加代理。

2、限制访问频率源码分析

self.check_throttles(request)

ottles(request)

ottles(self,request): ottled. ottled. throttle ottles(): 一个throttle对象，执行allow_request方法 ottle.allow_request(request,self): ottled(request,throttle.wait()) ottle.wait()表示还要等多少秒就能访问了

ottles

ottles(self): ottles that this view uses. [throttle() throttle self.throttle_classes]

ottles

throttle_classes = api_settings.DEFAULT_THRottLE_CLASSES

ottle_classes

ottle(object): ottling of requests. NotImplementedError( 0. If not use all of xff = request.Meta.get( remote_addr = request.Meta.get( num_proxies = num_proxies num_proxies == 0 xff addrs = xff.split( client_addr = addrs[- .join(xff.split()) xff None

ottle

display: none;">

zz

display: inline;">也可以重写allow_request方法

ottled(self,wait): ottled,determine what kind of exception to raise. exceptions.Throttled(wait)

错误信息throttled

ottled(APIException): status_code = default_detail = _(ottled. extra_detail_singular = extra_detail_plural = default_code = ottled __init__(self,wait=None,detail=None,code= detail detail = wait wait = detail = force_text(ungettext(self.extra_detail_singular.format(wait= self.extra_detail_plural.format(wait= self.wait = super(Throttled,self).__init__(detail,code)

ottled(wait)错误信息详情

3.方法

a. 基于用户IP限制访问频率

django.conf.urls

rest_framework.views rest_framework.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):
ctime = time.time

</span><span style="color: #0000ff;"&gt;def</span><span style="color: #000000;"&gt; get_ident(self,request):
    </span><span style="color: #800000;"&gt;"""</span><span style="color: #800000;"&gt;
    根据用户IP和代理IP，当做请求者的唯一IP
    Identify the machine making the request by parsing HTTP_X_FORWARDED_FOR
    if present and number of proxies is > 0. If not use all of
    HTTP_X_FORWARDED_FOR if it is available,if not use REMOTE_ADDR.
    </span><span style="color: #800000;"&gt;"""</span><span style="color: #000000;"&gt;
    xff </span>= request.<a href="/tag/Meta/" target="_blank" class="keywords">Meta</a>.get(<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;HTTP_X_FORWARDED_FOR</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;)
    remote_addr </span>= request.<a href="/tag/Meta/" target="_blank" class="keywords">Meta</a>.get(<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;REMOTE_ADDR</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;)
    num_proxies </span>=<span style="color: #000000;"&gt; api_settings.NUM_PROXIES

    </span><span style="color: #0000ff;"&gt;if</span> num_proxies <span style="color: #0000ff;"&gt;is</span> <span style="color: #0000ff;"&gt;not</span><span style="color: #000000;"&gt; None:
        </span><span style="color: #0000ff;"&gt;if</span> num_proxies == 0 <span style="color: #0000ff;"&gt;or</span> xff <span style="color: #0000ff;"&gt;is</span><span style="color: #000000;"&gt; None:
            </span><span style="color: #0000ff;"&gt;return</span><span style="color: #000000;"&gt; remote_addr
        addrs </span>= xff.split(<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;,</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;)
        client_addr </span>= addrs[-<span style="color: #000000;"&gt;min(num_proxies,len(addrs))]
        </span><span style="color: #0000ff;"&gt;return</span><span style="color: #000000;"&gt; client_addr.strip()

    </span><span style="color: #0000ff;"&gt;return</span> <span style="color: #800000;"&gt;''</span>.join(xff.split()) <span style="color: #0000ff;"&gt;if</span> xff <span style="color: #0000ff;"&gt;else</span><span style="color: #000000;"&gt; remote_addr

</span><span style="color: #0000ff;"&gt;def</span><span style="color: #000000;"&gt; allow_request(self,view):
    </span><span style="color: #800000;"&gt;"""</span><span style="color: #800000;"&gt;
    是否仍然在允许范围内
    Return `True` if the request should be allowed,`False` otherwise.
    :p<a href="/tag/ara/" target="_blank" class="keywords">ara</a>m request: 
    :p<a href="/tag/ara/" target="_blank" class="keywords">ara</a>m view: 
    :return: True，表示可以通过；False表示已超过限制，不允许访问
    </span><span style="color: #800000;"&gt;"""</span>
    <span style="color: #008000;"&gt;#</span><span style="color: #008000;"&gt; <a href="/tag/huoqu/" target="_blank" class="keywords">获取</a><a href="/tag/yonghu/" target="_blank" class="keywords">用户</a>唯一标识（如：IP）</span>

    <span style="color: #008000;"&gt;#</span><span style="color: #008000;"&gt; 允许一分钟访问10次</span>
    num_request = 10<span style="color: #000000;"&gt;
    time_request </span>= 60<span style="color: #000000;"&gt;

    <a href="/tag/Now/" target="_blank" class="keywords">Now</a> </span>=<span style="color: #000000;"&gt; self.ctime()
    ident </span>=<span style="color: #000000;"&gt; self.get_ident(request)
    self.ident </span>=<span style="color: #000000;"&gt; ident
    </span><span style="color: #0000ff;"&gt;if</span> ident <span style="color: #0000ff;"&gt;not</span> <span style="color: #0000ff;"&gt;in</span><span style="color: #000000;"&gt; RECORD:
        RECORD[ident] </span>=<span style="color: #000000;"&gt; [<a href="/tag/Now/" target="_blank" class="keywords">Now</a>,]
        </span><span style="color: #0000ff;"&gt;return</span><span style="color: #000000;"&gt; True
    history </span>=<span style="color: #000000;"&gt; RECORD[ident]
    </span><span style="color: #0000ff;"&gt;while</span> history <span style="color: #0000ff;"&gt;and</span> history[-1] <= <a href="/tag/Now/" target="_blank" class="keywords">Now</a> -<span style="color: #000000;"&gt; time_request:
        history.pop()
    </span><span style="color: #0000ff;"&gt;if</span> len(history) <<span style="color: #000000;"&gt; num_request:
        history.insert(0,<a href="/tag/Now/" target="_blank" class="keywords">Now</a>)
        </span><span style="color: #0000ff;"&gt;return</span><span style="color: #000000;"&gt; True

</span><span style="color: #0000ff;"&gt;def</span><span style="color: #000000;"&gt; wait(self):
    </span><span style="color: #800000;"&gt;"""</span><span style="color: #800000;"&gt;
    多少秒后可以允许继续访问
    Optionally,return a recommended number of seconds to wait before
    the next request.
    </span><span style="color: #800000;"&gt;"""</span><span style="color: #000000;"&gt;
    last_time </span>=<span style="color: #000000;"&gt; RECORD[self.ident][0]
    <a href="/tag/Now/" target="_blank" class="keywords">Now</a> </span>=<span style="color: #000000;"&gt; self.ctime()
    </span><span style="color: #0000ff;"&gt;return</span> int(60 + last_time -<span style="color: #000000;"&gt; <a href="/tag/Now/" target="_blank" class="keywords">Now</a>)

class TestView(APIView):
throttle_classes = [TestThrottle,**kwargs):
return Response('PUT请求，响应内容')

</span><span style="color: #0000ff;"&gt;def</span><span style="color: #000000;"&gt; thr<a href="/tag/ott/" target="_blank" class="keywords">ott</a>led(self,wait):
    </span><span style="color: #800000;"&gt;"""</span><span style="color: #800000;"&gt;
    访问<a href="/tag/cishu/" target="_blank" class="keywords">次数</a>被限制时，定制<a href="/tag/cuowu/" target="_blank" class="keywords">错误</a>信息
    </span><span style="color: #800000;"&gt;"""</span>

    <span style="color: #0000ff;"&gt;class</span><span style="color: #000000;"&gt; Thr<a href="/tag/ott/" target="_blank" class="keywords">ott</a>led(exceptions.Thr<a href="/tag/ott/" target="_blank" class="keywords">ott</a>led):
        default_detail </span>= <span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;请求被限制.</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;
        extra_detail_singular </span>= <span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;请 {wait} 秒之后再重试.</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;
        extra_detail_plural </span>= <span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;请 {wait} 秒之后再重试.</span><span style="color: #800000;"&gt;'</span>

    <span style="color: #0000ff;"&gt;raise</span> Thr<a href="/tag/ott/" target="_blank" class="keywords">ott</a>led(wait)</pre>

b. 基于用户IP显示访问频率（利于Django缓存）

REST_FRAMEWORK =ottLE_RATEScope:

django.conf.urls

50207580-5e29-450b-93ee-d9862cce27e1')">

50207580-5e29-450b-93ee-d9862cce27e1" class="cnblogs_code_hide">


 rest_framework.views  rest_framework.response <span style="color: #0000ff;">from rest_framework <span style="color: #0000ff;">import<span style="color: #000000;"> exceptions

<span style="color: #0000ff;">from rest_framework.throttling <span style="color: #0000ff;">import<span style="color: #000000;"> SimpleRateThrottle
<span style="color: #0000ff;">class<span style="color: #000000;"> TestThrottle(SimpleRateThrottle):
</span><span style="color: #008000;"&gt;#</span><span style="color: #008000;"&gt; <a href="/tag/peizhiwenjian/" target="_blank" class="keywords">配置文件</a>定义的<a href="/tag/xianshi/" target="_blank" class="keywords">显示</a>频率的Key</span>
s<a href="/tag/cop/" target="_blank" class="keywords">cop</a>e = <span style="color: #800000;"&gt;"</span><span style="color: #800000;"&gt;test_s<a href="/tag/cop/" target="_blank" class="keywords">cop</a>e</span><span style="color: #800000;"&gt;"</span>

<span style="color: #0000ff;"&gt;def</span><span style="color: #000000;"&gt; get_cache_key(self,view):
    </span><span style="color: #800000;"&gt;"""</span><span style="color: #800000;"&gt;
    Should return a unique cache-key which can be used for thr<a href="/tag/ott/" target="_blank" class="keywords">ott</a>ling.
    Must be overridden.

    May return `None` if the request should not be thr<a href="/tag/ott/" target="_blank" class="keywords">ott</a>led.
    </span><span style="color: #800000;"&gt;"""</span>
    <span style="color: #0000ff;"&gt;if</span> <span style="color: #0000ff;"&gt;not</span><span style="color: #000000;"&gt; request.user:
        ident </span>=<span style="color: #000000;"&gt; self.get_ident(request)
    </span><span style="color: #0000ff;"&gt;else</span><span style="color: #000000;"&gt;:
        ident </span>=<span style="color: #000000;"&gt; request.user

    </span><span style="color: #0000ff;"&gt;return</span> self.cache_format %<span style="color: #000000;"&gt; {
        </span><span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;s<a href="/tag/cop/" target="_blank" class="keywords">cop</a>e</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;: self.s<a href="/tag/cop/" target="_blank" class="keywords">cop</a>e,</span><span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;ident</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;: ident
    }
<span style="color: #0000ff;">class<span style="color: #000000;"> TestView(APIView):

throttle_classes =<span style="color: #000000;"> [TestThrottle,wait):

<span style="color: #800000;">"""<span style="color: #800000;">

访问次数被限制时，定制错误信息

<span style="color: #800000;">"""
    <span style="color: #0000ff;"&gt;class</span><span style="color: #000000;"&gt; Thr<a href="/tag/ott/" target="_blank" class="keywords">ott</a>led(exceptions.Thr<a href="/tag/ott/" target="_blank" class="keywords">ott</a>led):
        default_detail </span>= <span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;请求被限制.</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;
        extra_detail_singular </span>= <span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;请 {wait} 秒之后再重试.</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;
        extra_detail_plural </span>= <span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;请 {wait} 秒之后再重试.</span><span style="color: #800000;"&gt;'</span>

    <span style="color: #0000ff;"&gt;raise</span> Thr<a href="/tag/ott/" target="_blank" class="keywords">ott</a>led(wait)</pre>

c. view中限制请求频率

REST_FRAMEWORK =ottLE_RATES:

django.conf.urls

# 继承 ScopedrateThrottle
class TestThrottle(ScopedrateThrottle):

</span><span style="color: #0000ff;"&gt;def</span><span style="color: #000000;"&gt; get_cache_key(self,]

</span><span style="color: #008000;"&gt;#</span><span style="color: #008000;"&gt; 在settings中<a href="/tag/huoqu/" target="_blank" class="keywords">获取</a> xxxxxx 对应的<a href="/tag/pinlvxianzhi/" target="_blank" class="keywords">频率限制</a>值</span>
thr<a href="/tag/ott/" target="_blank" class="keywords">ott</a>le_s<a href="/tag/cop/" target="_blank" class="keywords">cop</a>e = <span style="color: #800000;"&gt;"</span><span style="color: #800000;"&gt;xxxxxx</span><span style="color: #800000;"&gt;"</span>

<span style="color: #0000ff;"&gt;def</span> get(self,wait):
    </span><span style="color: #800000;"&gt;"""</span><span style="color: #800000;"&gt;
    访问<a href="/tag/cishu/" target="_blank" class="keywords">次数</a>被限制时，定制<a href="/tag/cuowu/" target="_blank" class="keywords">错误</a>信息
    </span><span style="color: #800000;"&gt;"""</span>

    <span style="color: #0000ff;"&gt;class</span><span style="color: #000000;"&gt; Thr<a href="/tag/ott/" target="_blank" class="keywords">ott</a>led(exceptions.Thr<a href="/tag/ott/" target="_blank" class="keywords">ott</a>led):
        default_detail </span>= <span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;请求被限制.</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;
        extra_detail_singular </span>= <span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;请 {wait} 秒之后再重试.</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;
        extra_detail_plural </span>= <span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;请 {wait} 秒之后再重试.</span><span style="color: #800000;"&gt;'</span>

    <span style="color: #0000ff;"&gt;raise</span> Thr<a href="/tag/ott/" target="_blank" class="keywords">ott</a>led(wait)</pre>

d. 匿名时用IP限制+登录时用Token限制

REST_FRAMEWORK =ottLE_RATES: :

django.conf.urls web.views.s3_throttling urlpatterns = [
url(r'^test/',]

rest_framework.views rest_framework.response from rest_framework.throttling import SimpleRateThrottle

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

<span style="color: #0000ff;"&gt;def</span><span style="color: #000000;"&gt; get_cache_key(self,view):
    </span><span style="color: #008000;"&gt;#</span><span style="color: #008000;"&gt; <a href="/tag/yonghu/" target="_blank" class="keywords">用户</a>已<a href="/tag/denglu/" target="_blank" class="keywords">登录</a>，则跳过 匿名<a href="/tag/pinlvxianzhi/" target="_blank" class="keywords">频率限制</a></span>
    <span style="color: #0000ff;"&gt;if</span><span style="color: #000000;"&gt; request.user:
        </span><span style="color: #0000ff;"&gt;return</span><span style="color: #000000;"&gt; None

    </span><span style="color: #0000ff;"&gt;return</span> self.cache_format %<span style="color: #000000;"&gt; {
        </span><span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;s<a href="/tag/cop/" target="_blank" class="keywords">cop</a>e</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;: self.s<a href="/tag/cop/" target="_blank" class="keywords">cop</a>e,</span><span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;ident</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;: self.get_ident(request)
    }

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

<span style="color: #0000ff;"&gt;def</span><span style="color: #000000;"&gt; get_ident(self,request):
    </span><span style="color: #800000;"&gt;"""</span><span style="color: #800000;"&gt;
    认证成功时：request.user是<a href="/tag/yonghu/" target="_blank" class="keywords">用户</a>对象；request.auth是token对象
    :p<a href="/tag/ara/" target="_blank" class="keywords">ara</a>m request: 
    :return: 
    </span><span style="color: #800000;"&gt;"""</span>
    <span style="color: #008000;"&gt;#</span><span style="color: #008000;"&gt; return request.auth.token</span>
    <span style="color: #0000ff;"&gt;return</span> <span style="color: #800000;"&gt;"</span><span style="color: #800000;"&gt;user_token</span><span style="color: #800000;"&gt;"</span>

<span style="color: #0000ff;"&gt;def</span><span style="color: #000000;"&gt; get_cache_key(self,view):
    </span><span style="color: #800000;"&gt;"""</span><span style="color: #800000;"&gt;
    <a href="/tag/huoqu/" target="_blank" class="keywords">获取</a>缓存key
    :p<a href="/tag/ara/" target="_blank" class="keywords">ara</a>m request: 
    :p<a href="/tag/ara/" target="_blank" class="keywords">ara</a>m view: 
    :return: 
    </span><span style="color: #800000;"&gt;"""</span>
    <span style="color: #008000;"&gt;#</span><span style="color: #008000;"&gt; 未<a href="/tag/denglu/" target="_blank" class="keywords">登录</a><a href="/tag/yonghu/" target="_blank" class="keywords">用户</a>，则跳过 Token限制</span>
    <span style="color: #0000ff;"&gt;if</span> <span style="color: #0000ff;"&gt;not</span><span style="color: #000000;"&gt; request.user:
        </span><span style="color: #0000ff;"&gt;return</span><span style="color: #000000;"&gt; None

    </span><span style="color: #0000ff;"&gt;return</span> self.cache_format %<span style="color: #000000;"&gt; {
        </span><span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;s<a href="/tag/cop/" target="_blank" class="keywords">cop</a>e</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;: self.s<a href="/tag/cop/" target="_blank" class="keywords">cop</a>e,</span><span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;ident</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;: self.get_ident(request)
    }

class TestView(APIView):
throttle_classes = [LuffyUserRateThrottle,LuffyAnonRateThrottle,**kwargs):
return Response('PUT请求，响应内容')

e. 全局使用

REST_FRAMEWORK =ottLE_CLASSESottles.throttles.LuffyAnonRateThrottleottles.throttles.LuffyUserRateThrottleottLE_RATES: : :

下面来看看最简单的从源码中分析的示例,这只是举例说明了一下

django.conf.urls app04 urlpatterns = url( ]

django.shortcuts rest_framework.views rest_framework.response rest_framework ottle(object): 1000 authentication_classes = [] 用户 permission_classes = [] throttle_classes =ottle,] dispatch Response( ottled(self,wait): 方法设置中文错误 ottled(wait) ottle(exceptions.Throttled): default_detail = extra_detail_singular = extra_detail_plural = default_code = MyThrottle(wait)

需求：对匿名用户进行限制，每个用户一分钟允许访问10次（只针对用户来说）

a、基于用户IP限制访问频率

流程分析：

先获取用户信息，如果是匿名用户，获取IP。如果不是匿名用户就可以获取用户名。
获取匿名用户IP，在request里面获取，比如IP= 1.1.1.1。
吧获取到的IP添加到到recode字典里面，需要在添加之前先限制一下。
如果时间间隔大于60秒，说明时间久远了，就把那个时间给剔除了pop。在timelist列表里面现在留的是有效的访问时间段。
然后判断他的访问次数超过了10次没有，如果超过了时间就return False。
美中不足的是时间是固定的，我们改变他为动态的：列表里面最开始进来的时间和当前的时间进行比较，看需要等多久。

具体实现：

django.shortcuts rest_framework.views rest_framework.response rest_framework rest_framework.throttling BaseThrottle,SimpleRateThrottle RECORD = ottle(BaseThrottle): 匿名用户进行限制，每个用户一分钟访问10次 ctime = ip = ip RECORD[ip] = time_list = RECORD[ip] 获取ip里面的值 val = time_list[-1]一个时间，也就是访问最早的时间 (ctime-60)>val: 次数是不是超过10次 len(time_list) >10 False True ctime = first_in_time = RECORD[][-1 wt = 60-(ctime-irst_in_time) authentication_classes = [] 用户 permission_classes = [] throttle_classes =ottle,] dispatch Response( ottled(self,wait): 方法设置中文错误 ottled(wait) ottle(exceptions.Throttled): default_detail = extra_detail_singular = extra_detail_plural = default_code = MyThrottle(wait)

ottling import BaseThrottle,SimpleRateThrottle #限制访问频率 ottle(BaseThrottle): 匿名用户进行限制，每个用户一分钟访问10次 ''' 获取ip里面的值一个时间，也就是访问最早的时间 val: #吧时间大于60秒的给剔除了次数是不是超过10次 10: irst_in_time = RECORD['1.1.1.1'][-1] irst_in_time) 用户 ottle_classes = [MyThrottle,] dispatch ottled(self,wait): 方法设置中文错误''' ottled(wait) ottle(exceptions.Throttled): ottle(wait) django.shortcuts rest_framework.views rest_framework.response rest_framework rest_framework.throttling BaseThrottle,SimpleRateThrottle RECORD = ottle(BaseThrottle): 匿名用户进行限制，每个用户一分钟访问10次 ctime = self.ip = self.ip RECORD[self.ip] = time_list = RECORD[self.ip] 获取ip里面的值 val = time_list[-1]一个时间，也就是访问最早的时间 (ctime-60)>val: 次数是不是超过10次 len(time_list) >10 False True ctime = first_in_time = RECORD[self.ip][-1 wt = 60-(ctime-irst_in_time) authentication_classes = [] 用户 permission_classes = [] throttle_classes =ottle,] dispatch Response( ottled(self,wait): 方法设置中文错误 ottled(wait) ottle(exceptions.Throttled): default_detail = extra_detail_singular = extra_detail_plural = default_code = MyThrottle(wait)

b、用resetframework内部的限制访问频率（利于Django缓存）

源码分析：

rest_framework.throttling BaseThrottle,SimpleRateThrottle

ottle(object): ottling of requests. NotImplementedError( get_ident(self,request): 0. If not use all of xff = request.Meta.get( remote_addr = request.Meta.get() 获取IP等 num_proxies = num_proxies num_proxies == 0 xff addrs = xff.split( None

ottle相当于一个抽象类

ottle(BaseThrottle): 一个简单的缓存实现，只需要` get_cache_key() `。被覆盖。属性设置的。类。该属性是一个字符串的形式number_of_requests /期。 IoUs request @R_902_4045@ion used for throttling is stored in the cache. cache = timer = cache_format = ottle_%(scope)s_%(ident)s scope = THRottLE_RATES =ottLE_RATES __init__ getattr(self, self.rate = self.num_requests,self.duration = get_cache_key(self,view):一个半成品，我们可以来补充它 ottling. ottled. NotImplementedError( mine the string representation of the allowed request rate. getattr(self,cope msg = (cope` or `.rate` for '%s' throttle % self..__name__ figured(msg) ottLE_RATES[self.scope] msg = ottle rate set for '%s' scope %cope figured(msg) , rate num,period = rate.split( num_requests = duration = {: 1,: 60,: 3600,: 86400 ottled. ottle_success`. ottle_failure`. self.rate self.key = self.get_cache_key(request,view) self.key self.history = self.cache.get(self.key,[]) 默认是一个列表，赋值给了self.history，一个ip对应的访问记录 self.Now = Now passed the ottle duration self.history self.history[-1] <= self.Now - len(self.history) >= ottle_failure() ottle_success() ottle_success(self): Now) ottle_failure(self): Failed due to throttling. remaining_duration = self.duration - (self.Now - self.history[-1 remaining_duration = available_requests = self.num_requests - len(self.history) + 1 available_requests <= remaining_duration / float(available_requests)

ottle

请求一进来会先执行SimpleRateThrottle这个类的构造方法

__init__ getattr(self,None): self.rate = self.get_rate() 一个scope ，2/m self.num_requests,self.duration = self.parse_rate(self.rate)

__init__

mine the string representation of the allowed request rate. getattr(self,cope,None): cope，没有就报错了 msg = (cope` or `.rate` for '%s' throttle % self..__name__ figured(msg) ottLE_RATES[self.scope] msg = ottle rate set for '%s' scope %cope ImproperlyConfigured(msg)

, rate num,period = rate.split( num_requests = duration = {: 1,: 86400 (num_requests,duration)

ottled. ottle_success`. ottle_failure`. self.rate self.key = self.get_cache_key(request,view) self.key True self.history = self.cache.get(self.key,[]) 默认是一个列表，赋值给了self.history，一个ip对应的访问记录 self.Now = Now passed the ottle duration self.history self.history[-1] <= self.Now - len(self.history) >= ottle_failure() self.throttle_success()

remaining_duration = self.duration - (self.Now - self.history[-1 remaining_duration = available_requests = self.num_requests - len(self.history) + 1 available_requests <= remaining_duration / float(available_requests)

代码实现：

ottle(SimpleRateThrottle): scope = self.get_ident(request) authentication_classes = [] 用户 permission_classes = [] throttle_classes =ottle,] dispatch Response( ottled(self,wait): 方法设置中文错误 ottled(wait) ottle(exceptions.Throttled): default_detail = extra_detail_singular = extra_detail_plural = default_code = MyThrottle(wait)

记得在settings里面配置

REST_FRAMEWORK = ottLE_RATES : 文件 CACHES = : : ,文件路径 }

对匿名用户进行限制，每个用户1分钟允许访问5次，对于登录的普通用户1分钟访问10次，VIP用户一分钟访问20次

比如首页可以匿名访问
#先认证，只有认证了才知道是不是匿名的，
#权限登录成功之后才能访问，，index页面就不需要权限了
If request.user #判断登录了没有

django.contrib django.conf.urls app05 urlpatterns = url( url( ]

django.shortcuts rest_framework.views rest_framework.response rest_framework.authentication BaseAuthentication rest_framework.throttling BaseThrottle,SimpleRateThrottle rest_framework.permissions rest_framework app01 用户是否存在,如果存在就返回user和auth，如果没有就返回 token = request.query_params.get( obj = models.UserInfo.objects.filter(token=irst() None message= True False message = request.user== True False ottle(SimpleRateThrottle): scope = 次数和时间 None 登录用户我不管匿名用户 self.get_ident(request) 一个唯一标识IP ottle(SimpleRateThrottle): scope = 登录用户 None 匿名用户我不管首页支持匿名访问, 登录就可以访问 authentication_classes = [MyAuthentcate,] 匿名用户 permission_classes = [] throttle_classes = [AnonThrottle,UserThrottle,] 匿名用户和普通用户的访问限制 dispatch Response(首页 ottled(self,wait): 方法设置中文错误 ottled(wait) ottle(exceptions.Throttled): default_detail = extra_detail_singular = extra_detail_plural = default_code = ottle(wait) 登录就可以访问 authentication_classes = [MyAuthentcate,] 匿名用户 permission_classes = [MyPermission,] throttle_classes = [AnonThrottle,] 匿名用户和普通用户的访问限制 dispatch Response(页面 ottled(self,wait): 方法设置中文错误 ottled(wait) ottle(exceptions.Throttled): default_detail = extra_detail_singular = extra_detail_plural = default_code = MyThrottle(wait)

1、认证：就是检查用户是否存在；如果存在返回(request.user,request.auth)；不存在request.user/request.auth=None

2、权限：进行职责的划分

3、限制访问频率

- 类：authenticate/authenticate_header -- - - --=-= 权限
- 类：has_permission/has_object_permission
- 返回值：
- True、#有权限

False、#无权限

exceptions.PermissionDenied(detail="错误信息") #异常自己随意，想抛就抛，错误信息自己指定
- 配置：
- 视图：
class IndexView(APIView):
permission_classes = [MyPermission,]
- 全局：
REST_FRAMEWORK = {
"DEFAULT_PERMISSION_CLASSES": [
# "app02.utils.MyAuthentication",}
限流
- 类：allow_request/wait PS: scope = "wdp_user"
- 返回值：
　　　　　　return True、#不限制
　　　　　　return False #限制
- 配置：
- 视图：
class IndexView(APIView):

      thr<a href="/tag/ott/" target="_blank" class="keywords">ott</a>le_classes</span>=<span style="color: #000000;"&gt;[AnonThr<a href="/tag/ott/" target="_blank" class="keywords">ott</a>le,]
      </span><span style="color: #0000ff;"&gt;def</span> get(self,**<span style="color: #000000;"&gt;<a href="/tag/kwargs/" target="_blank" class="keywords">kwargs</a>):
          self.<a href="/tag/dis/" target="_blank" class="keywords">dis</a>patch
          </span><span style="color: #0000ff;"&gt;return</span> Response(<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;访问<a href="/tag/shouye/" target="_blank" class="keywords">首页</a></span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;)

- 全局
REST_FRAMEWORK = {
"DEFAULT_THRottLE_CLASSES":[

      ],</span><span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;DEFAULT_THR<a href="/tag/ott/" target="_blank" class="keywords">ott</a>LE_RATES</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;:{
          </span><span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;wdp_anon</span><span style="color: #800000;"&gt;'</span>:<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;5/minute</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;,</span><span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;wdp_user</span><span style="color: #800000;"&gt;'</span>:<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;10/minute</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;,}
  }

Django Rest Framework(认证、权限、限制访问频率)

一、认证和授权

用户访问次数/频率限制

1、为什么要限流呢

2、限制访问频率源码分析

3.方法

相关推荐