Django   发布时间:2022-04-10  发布网站:大佬教程  code.js-code.com
大佬教程收集整理的这篇文章主要介绍了Django Rest Framework(认证、权限、限制访问频率)大佬教程大佬觉得挺不错的,现在分享给大家,也给大家做个参考。

<div id="cnblogs_posT_Body" class="blogpost-body">

一、认证和授权

a. 用户url传入的token认证

@H_772_10@

django.conf.urls urlpatterns =<span style="color: #000000;"> [
url(r
<span style="color: #800000;">'
<span style="color: #800000;">^test/
<span style="color: #800000;">'
<span style="color: #000000;">,TestView.as_view()),]

s.py

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

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

token_list =<span style="color: #000000;"> [
<span style="color: #800000;">'<span style="color: #800000;">sfsfss123kuf3j123<span style="color: #800000;">'<span style="color: #000000;">,<span style="color: #800000;">'<span style="color: #800000;">asijnfowerkkf9812<span style="color: #800000;">'<span style="color: #000000;">,]

<span style="color: #0000ff;">class<span style="color: #000000;"> TestAuthentication(BaseAuthentication):
<span style="color: #0000ff;">def<span style="color: #000000;"> authenticate(self,request):
<span style="color: #800000;">"""<span style="color: #800000;">
用户认证,如果验证成功后返回元组: (用户,用户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()# 默认值为:None
            else:
                self.auth = None
        (user,token)表示验证通过并设置用户名和Token;
        AuthenticationFailed异常
    </span><span style="color: #800000;"&gt;"""</span><span style="color: #000000;"&gt;
    val </span>= request.query_params.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.AuthenticationFailed(<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;return</span> (<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;登录用户</span><span style="color: #800000;"&gt;'</span>,<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;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 scheR_170_11845@e 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 style="color: #0000ff;">class<span style="color: #000000;"> TestView(APIView):
<span style="color: #008000;">#<span style="color: #008000;"> 认证的动作是由request.user触发
authentication_classes =<span style="color: #000000;"> [TestAuthentication,]

</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;kwargs):
    </span><span style="color: #008000;"&gt;#</span><span style="color: #008000;"&gt; self.dispatch</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请求,响应内容</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;kwargs):
    </span><span style="color: #0000ff;"&gt;return</span> Response(<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;POST请求,响应内容</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;kwargs):
    </span><span style="color: #0000ff;"&gt;return</span> Response(<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;PUT请求,响应内容</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;)

views.py

s.py

b. 请求头认证

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

django.conf.urls
s.py

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

rest_framework.views rest_framework.response rest_framework.authentication rest_framework.request request rest_framework token_list =<span style="color: #000000;"> [
<span style="color: #800000;">'
<span style="color: #800000;">sfsfss123kuf3j123
<span style="color: #800000;">'
<span style="color: #000000;">,表示跳过该验证;
如果跳过了所有认证,默认用户和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)表示验证通过并设置用户名和Token;
        AuthenticationFailed异常
    </span><span style="color: #800000;"&gt;"""</span>
    <span style="color: #0000ff;"&gt;import</span><span style="color: #000000;"&gt; base64
    auth </span>= request.META.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.AuthenticationFailed(<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.AuthenticationFailed(<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;登录用户</span><span style="color: #800000;"&gt;'</span>,<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;else</span><span style="color: #000000;"&gt;:
        </span><span style="color: #0000ff;"&gt;raise</span> exceptions.AuthenticationFailed(<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;def</span><span style="color: #000000;"&gt; authenticate_header(self,or `None` if the
    authentication scheR_170_11845@e 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>

<span style="color: #0000ff;">class<span style="color: #000000;"> TestView(APIView):
authentication_classes =<span style="color: #000000;"> [TestAuthentication,]
permission_classes =<span style="color: #000000;"> []

</span><span style="color: #0000ff;"&gt;def</span> get(self,**<span style="color: #000000;"&gt;kwargs):
    </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请求,响应内容</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;kwargs):
    </span><span style="color: #0000ff;"&gt;return</span> Response(<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;PUT请求,响应内容</span><span style="color: #800000;"&gt;'</span>)</pre>

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

c.多个认证规则

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

django.conf.urls web.views.s2_auth urlpatterns =<span style="color: #000000;"> [
url(r
<span style="color: #800000;">'
<span style="color: #800000;">^test/
<span style="color: #800000;">'
<span style="color: #000000;">,]

s.py

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

rest_framework.views rest_framework.response rest_framework.authentication rest_framework.request request rest_framework token_list =<span style="color: #000000;"> [
<span style="color: #800000;">'
<span style="color: #800000;">sfsfss123kuf3j123
<span style="color: #800000;">'
<span style="color: #000000;">,]

<span style="color: #0000ff;">class<span style="color: #000000;"> Test1Authentication(BaseAuthentication):
<span style="color: #0000ff;">def<span style="color: #000000;"> 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()# 默认值为: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.AuthenticationFailed(<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.AuthenticationFailed(<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 scheR_170_11845@e 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>

<span style="color: #0000ff;">class<span style="color: #000000;"> Test2Authentication(BaseAuthentication):
<span style="color: #0000ff;">def<span style="color: #000000;"> authenticate(self,or None if the
authentication scheR_170_11845@e should return 403 Permission Denied responses.
<span style="color: #800000;">"""
<span style="color: #0000ff;">pass

<span style="color: #0000ff;">class<span style="color: #000000;"> TestView(APIView):
authentication_classes =<span style="color: #000000;"> [Test1Authentication,Test2Authentication]
permission_classes =<span style="color: #000000;"> []

</span><span style="color: #0000ff;"&gt;def</span> get(self,**<span style="color: #000000;"&gt;kwargs):
    </span><span style="color: #0000ff;"&gt;return</span> Response(<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;PUT请求,响应内容</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;)

views.py

s.py

d.认证和权限

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

django.conf.urls web.views urlpatterns =<span style="color: #000000;"> [
url(r
<span style="color: #800000;">'
<span style="color: #800000;">^test/
<span style="color: #800000;">'
<span style="color: #000000;">,]

s.py

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

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

token_list =<span style="color: #000000;"> [
<span style="color: #800000;">'<span style="color: #800000;">sfsfss123kuf3j123<span style="color: #800000;">'<span style="color: #000000;">,or None if the
authentication scheR_170_11845@e should return 403 Permission Denied responses.
<span style="color: #800000;">"""
<span style="color: #0000ff;">pass

<span style="color: #0000ff;">class<span style="color: #000000;"> TestPermission(BasePermission):
message = <span style="color: #800000;">"<span style="color: #800000;">权限验证失败<span style="color: #800000;">"

<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.
    :param request: 
    :param 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;管理员</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时调用</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时获取对象时,触发单独对象权限验证
    Return `True` if permission is granted,`false` otherwise.
    :param request: 
    :param view: 
    :param 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;管理员</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 style="color: #0000ff;">class<span style="color: #000000;"> TestView(APIView):
<span style="color: #008000;">#<span style="color: #008000;"> 认证的动作是由request.user触发
authentication_classes =<span style="color: #000000;"> [TestAuthentication,**<span style="color: #000000;">kwargs):
<span style="color: #0000ff;">return Response(<span style="color: #800000;">'<span style="color: #800000;">PUT请求,响应内容<span style="color: #800000;">')

s.py

e.全局使用

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

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

REST_FRAMEWORK ={ s.TestAuthentications.TestPermission
Tings.py

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

django.conf.urls urls.py

s.py

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

rest_framework.views rest_framework.response <span style="color: #0000ff;">class<span style="color: #000000;"> TestView(APIView):

</span><span style="color: #0000ff;"&gt;def</span> get(self,**<span style="color: #000000;"&gt;kwargs):
    </span><span style="color: #0000ff;"&gt;return</span> Response(<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;PUT请求,响应内容</span><span style="color: #800000;"&gt;'</span>)</pre>
s.py

g.自定义认证工作

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

request): token = request.query_params.get() token == (,) APIException( authenticate_header(self,request): serview(APIView): authentication_classes = get(self,** request.user) request.auth) Response()

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

title">二、权限

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

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

app03 django.conf.urls urlpatterns = url(s.AuthView.as_view()), url(rs.HostView.as_view()), url(rS.UsersView.as_view()), url(rs.SalaryView.as_view()), ]
s.py

@H_801_693@

@H_616_698@

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 request): token = request.query_params.get( obj = modelS.UserInfo.objects.filter(token= obj : None request): ject): message = has_permission(self,view): request.user: True false ject): message = has_permission(self,view): request.user== True false false表示无权限 authentication_classes = [] request): 用户都能访问 #匿名用户的request.user = none authentication_classes = permission_classes = [] request): request.user) request.auth) Response( request.user里面有值 authentication_classes = permission_classes = request): (request.user, Response( permission_denied(self,message=E): request is not permitted,determine what kind of exception to raise. request.authenticators request.successful_authenticator: false了,就会报下面的这个异常了 exceptions.NotAuthenticated(detail= exceptions.PermissionDenied(detail=messagE)
s.py
4898-bdb1-beb5a942a1d0')">

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

4898-bdb1-beb5a942a1d0" class="cnblogs_code_hide">
      
     message =
     authentication_classes = [MyAuthentication,]  
     permission_classes = [MyPermission,AdminPermission,] 没有管理员的权限
     request):
          Response( 
      permission_denied(self,message=E):
         
 request is not permitted,determine what kind of exception to raise.
         
          request.authenticators  request.successful_authenticator:
             false了,就会报下面的这个异常了
              exceptions.NotAuthenticated(detail=          exceptions.PermissionDenied(detail=messagE)

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

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

check_permissions(self,request): check if the request should be permitted. Raises an appropriate exception if the request is not permitted. permission false,则抛出异常 request,self): self.permission_denied( request,message=getattr(permission,messageE) )
permission_denied(self,message=E): request is not permitted,determine what kind of exception to raise. request.authenticators request.successful_authenticator: false了,就会报下面的这个异常了 s.NotAuthenticated() exceptions.PermissionDenied(detail=messagE)

那么我们可以重写permission_denied这个方法,如下:

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

request.user里面有值 authentication_classes = permission_classes = request): Response( permission_denied(self,message=E): request is not permitted,determine what kind of exception to raise. request.authenticators request.successful_authenticator: false了,就会报下面的这个异常了 exceptions.NotAuthenticated(detail= exceptions.PermissionDenied(detail=messagE)
s.py

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

2.全局使用

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

0-9537-4fc4-bd28-7049759e4b15')">

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

0-9537-4fc4-bd28-7049759e4b15" class="cnblogs_code_hide">
 REST_FRAMEWORK ={
          : None,
              s.MyAuthentication               s.MyPermission,
  }
Tings.py
4890-a0bd-a0484ea5828d')">

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

4890-a0bd-a0484ea5828d" class="cnblogs_code_hide">
      authentication_classes = []  
 
     request):
           
 
       用户都能访问  #匿名用户的request.user = none
      
     authentication_classes =     permission_classes = []  
     request):
         request.user)
         request.auth)
          Response( 
      request.user里面有值
     authentication_classes =     permission_classes =     request):
         (request.user,          Response( 
      permission_denied(self,message=E):
         
 request is not permitted,determine what kind of exception to raise.
         
          request.authenticators  request.successful_authenticator:
             false了,就会报下面的这个异常了
              exceptions.NotAuthenticated(detail=          exceptions.PermissionDenied(detail=messagE)
 
 
      
     message =
     authentication_classes = [MyAuthentication,]  
     permission_classes = [MyPermission,] 没有管理员的权限
     request):
          Response( 
      permission_denied(self,message=E):
         
 request is not permitted,determine what kind of exception to raise.
         
          request.authenticators  request.successful_authenticator:
             false了,就会报下面的这个异常了
              exceptions.NotAuthenticated(detail=          exceptions.PermissionDenied(detail=messagE)
s.py

title">三、用户访问次数/频率限制

1、为什么要限流呢

答:

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

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

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

self.check_throttles(request)
check_throttles(request)

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

check_throttles(self,request): check if request should be throttled. request is throttled. throttle request方法 request: false,说明限制访问频率 request(request,self): request,throttle.wait())
check_throttles

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

s. [throttle() throttle self.throttle_classes]

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

throttle_classes = api_setTings.DEFAULT_THROTTLE_CLASSES

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

ject): requests. request(self,view): request should be allowed,`false` otherwise. NotImplementedError(request() must be overridden request): request by parsing http_X_FORWARDED_FOR 0. If not use all of http_X_FORWARDED_FOR if it is available,if not use REMOTE_ADDR. xff = request.META.get(http_X_FORWARDED_FOR remote_addr = request.META.get( num_proxies =Tings.NUM_PROXIES num_proxies num_proxies == 0 xff addrs = xff.split( client_addr = addrs[- client_addr.Strip() .join(xff.split()) xff number of seconds to wait before request. None

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

@H_317_1618@

request方法
674-07cf-405d-97b9-0dd00ef93e4a')">

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

674-07cf-405d-97b9-0dd00ef93e4a" class="cnblogs_code_hide">
            
 request is throttled,determine what kind of exception to raise.
         
          exceptions.Throttled(wait)

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

status_code =s.http_429_TOO_MANY_requESTS default_detail = _(request was throttled. extra_detail_singular = {wait} second. extra_detail_plural = {wait} seconds. default_code = (self,wait=None,detail=None,code=E): 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).(detail,codE)
s.Throttled(wait)错误信息详情

3.方法

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

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

django.conf.urls
s.py
450ca1')">

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

450ca1" 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;"> BaseThrottle
<span style="color: #0000ff;">from
rest_framework.setTings <span style="color: #0000ff;">import
<span style="color: #000000;"> api_setTings

<span style="color: #008000;">#<span style="color: #008000;"> 保存访问记录
RECORD =<span style="color: #000000;"> {
<span style="color: #800000;">'<span style="color: #800000;">用户IP<span style="color: #800000;">': [12312139,12312135,12312133<span style="color: #000000;">,]
}

<span style="color: #0000ff;">class<span style="color: #000000;"> TestThrottle(BaseThrottlE):
ctime =<span style="color: #000000;"> 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.META.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.META.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.
    :param request: 
    :param view: 
    :return: True,表示可以通过;false表示已超过限制,不允许访问
    </span><span style="color: #800000;"&gt;"""</span>
    <span style="color: #008000;"&gt;#</span><span style="color: #008000;"&gt; 获取用户唯一标识(如: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;

    now </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; [now,]
        </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] <= now -<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,now)
        </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]
    now </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; now)

<span style="color: #0000ff;">class<span style="color: #000000;"> TestView(APIView):
throttle_classes =<span style="color: #000000;"> [TestThrottle,**<span style="color: #000000;">kwargs):
<span style="color: #0000ff;">return Response(<span style="color: #800000;">'<span style="color: #800000;">PUT请求,响应内容<span style="color: #800000;">'<span style="color: #000000;">)

</span><span style="color: #0000ff;"&gt;def</span><span style="color: #000000;"&gt; throttled(self,wait):
    </span><span style="color: #800000;"&gt;"""</span><span style="color: #800000;"&gt;
    访问次数被限制时,定制错误信息
    </span><span style="color: #800000;"&gt;"""</span>

    <span style="color: #0000ff;"&gt;class</span><span style="color: #000000;"&gt; Throttled(exceptions.Throttled):
        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> Throttled(wait)</pre>
s.py

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

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

REST_FRAMEWORK ={ { test_scope:
Tings.py

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

django.conf.urls
s.py
450b-93ee-d9862cce27e1')">

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

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; 配置文件定义的显示频率的Key</span>
scope = <span style="color: #800000;"&gt;"</span><span style="color: #800000;"&gt;test_scope</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 throttling.
    Must be overridden.

    May return `None` if the request should not be throttled.
    </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;scope</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;: self.scope,</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; Throttled(exceptions.Throttled):
        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> Throttled(wait)</pre>
s.py

c. view中限制请求频率

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

REST_FRAMEWORK ={ { :
Tings.py

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

django.conf.urls
s.py

@H_801_1975@

@H_28_1980@

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;"> ScopedRateThrottle

<span style="color: #008000;">#<span style="color: #008000;"> 继承 ScopedRateThrottle
<span style="color: #0000ff;">class<span style="color: #000000;"> 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中获取 xxxxxx 对应的频率限制值</span>
throttle_scope = <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;
    访问次数被限制时,定制错误信息
    </span><span style="color: #800000;"&gt;"""</span>

    <span style="color: #0000ff;"&gt;class</span><span style="color: #000000;"&gt; Throttled(exceptions.Throttled):
        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> Throttled(wait)</pre>
s.py

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

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

REST_FRAMEWORK ={ { : :
Tings.py

@H_55_2058@

django.conf.urls web.views.s3_throttling urlpatterns =<span style="color: #000000;"> [
url(r
<span style="color: #800000;">'
<span style="color: #800000;">^test/
<span style="color: #800000;">'
<span style="color: #000000;">,]

s.py

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

rest_framework.views rest_framework.response <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;"> LuffyAnonRateThrottle(SimpleRateThrottlE):
<span style="color: #800000;">"""<span style="color: #800000;">
匿名用户,根据IP进行限制
<span style="color: #800000;">"""<span style="color: #000000;">
scope = <span style="color: #800000;">"<span style="color: #800000;">luffy_anon<span style="color: #800000;">"

<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; 用户已登录,则跳过 匿名频率限制</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;scope</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;: self.scope,</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)
    }

<span style="color: #0000ff;">class<span style="color: #000000;"> LuffyUserRateThrottle(SimpleRateThrottlE):
<span style="color: #800000;">"""<span style="color: #800000;">
登录用户,根据用户token限制
<span style="color: #800000;">"""<span style="color: #000000;">
scope = <span style="color: #800000;">"<span style="color: #800000;">luffy_user<span style="color: #800000;">"

<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是用户对象;request.auth是token对象
    :param 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;
    获取缓存key
    :param request: 
    :param view: 
    :return: 
    </span><span style="color: #800000;"&gt;"""</span>
    <span style="color: #008000;"&gt;#</span><span style="color: #008000;"&gt; 未登录用户,则跳过 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;scope</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;: self.scope,</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)
    }

<span style="color: #0000ff;">class<span style="color: #000000;"> TestView(APIView):
throttle_classes =<span style="color: #000000;"> [LuffyUserRateThrottle,LuffyAnonRateThrottle,**<span style="color: #000000;">kwargs):
<span style="color: #0000ff;">return Response(<span style="color: #800000;">'<span style="color: #800000;">PUT请求,响应内容<span style="color: #800000;">')

s.py

e. 全局使用

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

REST_FRAMEWORK ={ s.throttles.throttles.LuffyAnonRateThrottles.throttles.throttles.LuffyUserRateThrottle{ : : :
Tings

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

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

django.conf.urls app04 urlpatterns = url(s.LimitView.as_view()), ]
s.py
0-9ac5-bb826eaac449')">

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

@H_954_2262@

0-9ac5-bb826eaac449" class="cnblogs_code_hide">
  django.shortcuts   rest_framework.views   rest_framework.response   rest_framework  
 
 myThrottle(object):
     request(self,view):
         false,限制
         
         
               1000
 
 
      authentication_classes = []  
     permission_classes = []  
     throttle_classes =myThrottle,]
     request):
         
          Response( 
              
         s.Throttled(wait)
         myThrottle(exceptions.Throttled):
             default_detail = 
             extra_detail_singular = {wait} second.
             extra_detail_plural = {wait} seconds.
             default_code = {wait}秒
          myThrottle(wait)
s.py

需求:对匿名用户进行限制,每个用户一分钟允许访问10次(只针对用户来说

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

流程分析:

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

具体实现:

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

django.shortcuts rest_framework.views rest_framework.response rest_framework rest_framework.throttling BaseThrottle,SimpleRateThrottle RECORD ={} myThrottle(BaseThrottlE): request(self,view): ctime = ip = ip RECORD[ip] =ctime] time_list = RECORD[ip] val = time_list[-1] (ctime-60)>val: len(time_list) >10 false false,限制 ctimE) True ctime = first_in_time = RECORD[][-1 wt = 60-(ctime-E) authentication_classes = [] permission_classes = [] throttle_classes =myThrottle,] request): Response( s.Throttled(wait) myThrottle(exceptions.Throttled): default_detail = extra_detail_singular = {wait} second. extra_detail_plural = {wait} seconds. default_code = {wait}秒 myThrottle(wait)

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

{} myThrottle(BaseThrottlE): request(self,view): ctime = time.time() ctime] val: #吧时间大于60秒的给剔除了 10: false # 返回false,限制 ctimE) ctime = time.time() ctime-first_in_timE) myThrottle,] request): s.Throttled(wait) myThrottle(exceptions.Throttled): myThrottle(wait) django.shortcuts rest_framework.views rest_framework.response rest_framework rest_framework.throttling BaseThrottle,SimpleRateThrottle RECORD ={} myThrottle(BaseThrottlE): request(self,view): ctime = self.ip =request) self.ip RECORD[self.ip] =ctime] time_list = RECORD[self.ip] val = time_list[-1] (ctime-60)>val: len(time_list) >10 false false,限制 ctimE) True ctime = first_in_time = RECORD[self.ip][-1 wt = 60-(ctime-E) authentication_classes = [] permission_classes = [] throttle_classes =myThrottle,] request): Response( s.Throttled(wait) myThrottle(exceptions.Throttled): default_detail = extra_detail_singular = {wait} second. extra_detail_plural = {wait} seconds. default_code = {wait}秒 myThrottle(wait)
做了改动

@H_626_2922@

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

源码分析:

rest_framework.throttling BaseThrottle,SimpleRateThrottle

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

ject): requests. request(self,`false` otherwise. NotImplementedError(request() must be overridden get_ident(self,request): request by parsing http_X_FORWARDED_FOR 0. If not use all of http_X_FORWARDED_FOR if it is available,if not use REMOTE_ADDR. xff = request.META.get(http_X_FORWARDED_FOR remote_addr = request.META.get() num_proxies =Tings.NUM_PROXIES num_proxies num_proxies == 0 xff addrs = xff.split(number of seconds to wait before request. None
-1455-4de3-9913-7a4c4db2be25')">

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

-1455-4de3-9913-7a4c4db2be25" class="cnblogs_code_hide">
 E):
     
  number_of_requests /期。
   requires `.get_cache_key()`
  
 requests / seconds) is set by a `rate` attribute on the View
 s.  The attribute is a String of the form 'number_of_requests/period'.
 
  
 request information used for throttling is stored in the cache.
     
     cache =     timer =     cache_format = E)s_%(ident)s
     scope =     THROTTLE_RATES =Tings.DEFAULT_THROTTLE_RATES
 
                 getattr(self,E):
             self.rate =         self.num_requests,self.duration =E)
 
      get_cache_key(self,view):这个相当于是一个半成品,我们可以来补充它
         
   
 request should not be throttled.
         
          NotImplementedError( 
              
 String representation of the allowed request rate.
         
           getattr(self,E):
             msg = ( %
                    self..              
                                            msg =  %              
     E):
         
 request rate String,return a two tuple of:
 ,
         
          rate              E)
         num,period = rate.split(         num_requests =         duration = {: 1,: 60,: 3600,: 86400[0]]
         requests,duration)
 
     
     request(self,view):
         
 check to see if the request should be throttled.
 
 success calls `throttle_success`.
          
          self.rate               
         self.key = self.get_cache_key(request,view)  
          self.key               
         self.history = self.cache.get(self.key,[])  
                                                         
         self.now = 
         requests from the history which have now passed the
         
          self.history  self.historY[-1] <= self.now -           len(self.history) >=requests:
                      success()
 
     success(self):
         
 request's timestamp along with the key
          
            
              
 request to the API has failed due to throttling.
         
         false
 
              
 request time in seconds.
         
                      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)

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

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

getattr(self,NonE): self.rate = self.get_rate() self.num_requests,self.duration = self.parse_rate(self.ratE)

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

String representation of the allowed request rate. getattr(self,,NonE): msg = ( % self.. msg = % ImproperlyConfigured(msg)

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

E): request rate String,return a two tuple of: , rate E) num,period = rate.split( num_requests = duration = {: 1,: 86400[0]] (num_requests,duration)

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

request(self,view): check to see if the request should be throttled. success calls `throttle_success`. self.rate self.key = self.get_cache_key(request,view) self.key True self.history = self.cache.get(self.key,[]) self.now = requests from the history which have now passed the self.history self.historY[-1] <= self.now - len(self.history) >=requests: self.throttle_success()
request

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

request time in seconds. 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)

代码实现:

@H_939_3616@

内部的限制访问频率############## E): scope = self.get_ident(request) authentication_classes = [] permission_classes = [] throttle_classes = request): Response( s.Throttled(wait) myThrottle(exceptions.Throttled): default_detail = extra_detail_singular = {wait} second. extra_detail_plural = {wait} seconds. default_code = {wait}秒 myThrottle(wait)
s.py

记得在setTings里面配置

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

REST_FRAMEWORK ={ s.MyPermission",#设置路径, { :minute CACHES ={ { BACKEND: BACkends.filebased.FileBasedCache LOCATION: , }
Tings.py

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

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

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

django.contrib django.conf.urls app05 urlpatterns = url(s.IndexView.as_view()), url(s.ManageView.as_view()), ]
s.py

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

django.shortcuts rest_framework.views rest_framework.response rest_framework.authentication BaseAuthentication rest_framework.throttling BaseThrottle,SimpleRateThrottle rest_framework.permissions rest_framework app01 request): token = request.query_params.get( obj = modelS.UserInfo.objects.filter(token= None message= request.user: True false false表示无权限 message = request.user== True false false表示无权限 E): scope = request.user: None self.get_ident(request) E): scope = request.user: request.user None authentication_classes = [MyAuthentcate,] permission_classes = [] throttle_classes = [AnonThrottle,UserThrottle,] request): Response( s.Throttled(wait) myThrottle(exceptions.Throttled): default_detail = extra_detail_singular = {wait} second. extra_detail_plural = {wait} seconds. default_code = {wait}秒 myThrottle(wait) authentication_classes = [MyAuthentcate,] permission_classes = [MyPermission,] throttle_classes = [AnonThrottle,] request): Response( s.Throttled(wait) myThrottle(exceptions.Throttled): default_detail = extra_detail_singular = {wait} second. extra_detail_plural = {wait} seconds. default_code = {wait}秒 myThrottle(wait)
s.py

title">四、总结

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

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

3、限制访问频率

- 类:authenticate/authenticate_header -- - - --=-={ s.MyAuthentication", 权限
- 类:has_permission/<span style="color: #000000;">has_object_permission
-<span style="color: #000000;"> 返回值:
- True、<span style="color: #008000;">#
<span style="color: #008000;">有权限

  • false、<span style="color: #008000;">#<span style="color: #008000;">无权限

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

          throttle_classes</span>=<span style="color: #000000;"&gt;[AnonThrottle,]
          </span><span style="color: #0000ff;"&gt;def</span> get(self,**<span style="color: #000000;"&gt;kwargs):
              self.dispatch
              </span><span style="color: #0000ff;"&gt;return</span> Response(<span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;访问首页</span><span style="color: #800000;"&gt;'</span><span style="color: #000000;"&gt;)

    -<span style="color: #000000;"> 全局
    REST_FRAMEWORK =<span style="color: #000000;"> {
    <span style="color: #800000;">"<span style="color: #800000;">DEFAULT_THROTTLE_CLASSES<span style="color: #800000;">"<span style="color: #000000;">:[

          ],</span><span style="color: #800000;"&gt;'</span><span style="color: #800000;"&gt;DEFAULT_THROTTLE_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(认证、权限、限制访问频率)全部内容,希望文章能够帮你解决Django Rest Framework(认证、权限、限制访问频率)所遇到的程序开发问题。

如果觉得大佬教程网站内容还不错,欢迎将大佬教程推荐给程序员好友。

本图文内容来源于网友网络收集整理提供,作为学习参考使用,版权属于原作者。
如您有任何意见或建议可联系处理。小编QQ:384754419,请注明来意。
标签: