Django_rest_framework_組件（authentication、permission、throttle）

時間  2019-11-17
標籤 django rest framework 組件 authentication permission throttle 欄目 Python 简体版
原文   原文鏈接

 

 

認證組件

說明

from rest_framework.authentication import BaseAuthentication

class TestAuthentication(BaseAuthentication):
    def authenticate(self, request):
        """
        用戶認證，若是驗證成功後返回元組： (用戶,用戶Token)
        :param request: 
        :return:
                return1：(user,token)表示驗證經過並設置用戶名和Token；
                return2：AuthenticationFailed異常
                return3：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

        """
        ....return ('登陸用戶', '用戶token')

    def authenticate_header(self, request):
        """
        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.
        """
        pass

 

局部認證

在須要認證的視圖類里加上authentication_classes = [認證組件1類名,認證組件2類名....]html

示例以下：python

seralizers.pydjango

1
2
3
4
5
6
7
from  rest_framework  import  serializers
from  app01  import  models
 
class  PublishSerializers(serializers.ModelSerializer):
     class  Meta:
         model  =  models.Publish
         fields  =  '__all__'

auth.pyjson

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from  rest_framework.authentication  import  BaseAuthentication
from  rest_framework  import  exceptions
from  app01  import  models
 
class  TokenAuth(BaseAuthentication):
     def  authenticate( self ,request):
         '''函數名必須叫authenticate'''
         # 驗證條件根據需求設置（此示例爲須要有token值）
         token  =  request.GET.get( 'token' )
         token_obj  =  models.Token.objects. filter (token = token).first()
         if  not  token_obj:
             # 若是驗證失敗，須要跑出AuthenticationFailed錯誤
             raise  exceptions.AuthenticationFailed( "驗證失敗!" )
         else :
             user  =  token_obj.user
             # 若是驗證成功，須要返回一個元組，分別是用戶以及驗證類的實例對象，而後內部會賦值給request.user和request.auth
             return  user.username,token_obj 
from rest_framework.authentication import BaseAuthentication
from rest_framework.exceptions import AuthenticationFailed
from app01 import models
 
class BlackNameAuth(BaseAuthentication):
    '''黑名單認證'''
 
    def authenticate(self, request):
        BLACK_NAME_LIST = ['小花', '小翠']
 
        # 經過從url獲取user_id的方式模擬用戶登陸
        user_id = request.GET.get('uid')
        user = models.UserInfo.objects.filter(pk=user_id).first()
 
        if not user or user.username in BLACK_NAME_LIST:
            raise AuthenticationFailed('您沒有登陸或者被關小黑屋啦')
        else:
            return user.username,user_id
auth2

views.pyapi

1
2
3
4
5
6
7
8
9
10
from  rest_framework  import  viewsets
from  app01.auth  import  TokenAuth
 
class  PublishViewSet(viewsets.ModelViewSet):
     # 在這裏配置authentication_classes
     # 注意，值爲一個列表，能夠放多個認證組件類名  
     authentication_classes  =  [TokenAuth]
 
     queryset  =  models.Publish.objects. all ()
     serializer_class  =  serializer.PublishSerializers

全局認證

在setting.py裏配置以下：restful

1
2
3
REST_FRAMEWORK  =  {
     "DEFAULT_AUTHENTICATION_CLASSES" : [ "app01.auth.TokenAuth" ,]
}

　　這樣配置以後，每一個視圖類都要通過認證成功以後才能執行下一步，app

若是有某些方法不須要認證，如login函數，則須要在login函數中單獨加入一個配置屬性：dom

1
authentication_classes  =  []  #本身的類裏有的話就調用此類的配置，爲空既什麼都不作

 

權限組件

說明

class TestPermission(BasePermission):
    message = "權限驗證失敗"

    def has_permission(self, request, view):
        """
        判斷是否有權限訪問當前請求
        Return `True` if permission is granted, `False` otherwise.
        :param request: 
        :param view: 
        :return: True有權限；False無權限
        """
        if request.user == "管理員":
            return True

    # GenericAPIView中get_object時調用
    def has_object_permission(self, request, view, obj):
        """
        視圖繼承GenericAPIView，並在其中使用get_object時獲取對象時，觸發單獨對象權限驗證
        Return `True` if permission is granted, `False` otherwise.
        :param request: 
        :param view: 
        :param obj: 
        :return: True有權限；False無權限
        """
        if request.user == "管理員":
            return True

局部權限

permission.pyide

1
2
3
4
5
6
7
8
9
10
11
12
13
from  app01  import  models
class  VipPermission():
 
     def  has_permission( self ,request,view):
         # 通過認證組件以後將用戶名賦值給了request.user
         # 這裏能夠直接取到
         username  =  request.user
         user  =  models.User.objects. filter (username = username).first()
         # 若是用戶的vip值爲1，即爲True，則經過認證，不然return False
         if  user.vip:
             return  True
         else :
             return  False

views.py函數

1
2
3
4
5
6
7
8
9
from  rest_framework  import  viewsets
from  app01.auth  import  TokenAuth
from  app01.permission  import  VipPermission
class  PublishViewSet(viewsets.ModelViewSet):
     authentication_classes  =  [TokenAuth]
     permission_classes  =  [VipPermission]
 
     queryset  =  models.Publish.objects. all ()
     serializer_class  =  serializer.PublishSerializers

　　這個時候若是登陸用戶是vip，則會繼續執行下一步，若是是普通用戶，則會返回錯誤信息，以下：

1
{ "detail" : "You do not have permission to perform this action." }

　　若是須要自定義錯誤信息，只須要在類裏定義一個message屬性便可，以下：

1
message = "只有超級用戶才能訪問"

全局權限

1
2
3
4
5
6
REST_FRAMEWORK  =  {
     # 認證組件
     "DEFAULT_AUTHENTICATION_CLASSES" : [ "app01.auth.TokenAuth" ,],
     # 權限組件
     "DEFAULT_PERMISSION_CLASSES" : [ "app01.permission.VipPermission" ,],
}

throttle(訪問頻率)組件

局部視圖throttle

在app01.service.throttles.py中：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
from  rest_framework.throttling  import  BaseThrottle
 
VISIT_RECORD = {}
class  VisitThrottle(BaseThrottle):
 
     def  __init__( self ):
         self .history = None
 
     def  allow_request( self ,request,view):
         remote_addr  =  request.META.get( 'REMOTE_ADDR' )
         print (remote_addr)
         import  time
         ctime = time.time()
 
         if  remote_addr  not  in  VISIT_RECORD:
             VISIT_RECORD[remote_addr] = [ctime,]
             return  True
 
         history = VISIT_RECORD.get(remote_addr)
         self .history = history
 
         while  history  and  history[ - 1 ]<ctime - 60 :
             history.pop()
 
         if  len (history)< 3 :
             history.insert( 0 ,ctime)
             return  True
         else :
             return  False
 
     def  wait( self ):
         import  time
         ctime = time.time()
         return  60 - (ctime - self .history[ - 1 ])

在views.py中：

1
2
3
4
5
6
from  app01.service.throttles  import  *
 
class  BookViewSet(generics.ListCreateAPIView):
     throttle_classes  =  [VisitThrottle,]
     queryset  =  Book.objects. all ()
     serializer_class  =  BookSerializers

全局視圖throttle

1
2
3
4
5
REST_FRAMEWORK = {
     "DEFAULT_AUTHENTICATION_CLASSES" :[ "app01.service.auth.Authentication" ,],
     "DEFAULT_PERMISSION_CLASSES" :[ "app01.service.permissions.SVIPPermission" ,],
     "DEFAULT_THROTTLE_CLASSES" :[ "app01.service.throttles.VisitThrottle" ,]
}

內置throttle類

在app01.service.throttles.py修改成：

1
2
3
4
5
6
class  VisitThrottle(SimpleRateThrottle):
 
     scope = "visit_rate"
     def  get_cache_key( self , request, view):
 
         return  self .get_ident(request)

settings.py設置：

1
2
3
4
5
6
7
8
REST_FRAMEWORK = {
     "DEFAULT_AUTHENTICATION_CLASSES" :[ "app01.service.auth.Authentication" ,],
     "DEFAULT_PERMISSION_CLASSES" :[ "app01.service.permissions.SVIPPermission" ,],
     "DEFAULT_THROTTLE_CLASSES" :[ "app01.service.throttles.VisitThrottle" ,],
     "DEFAULT_THROTTLE_RATES" :{
         "visit_rate" : "5/m" ,  # 1分鐘5次，同理，5/s,5/h,5/d，表示秒/時/日
     }
}

views.py上可重寫throttled方法

#首頁支持匿名訪問,
#無須要登陸就能夠訪問
class IndexView(APIView):
    authentication_classes = [MyAuthentcate,]   #認證判斷他是否是匿名用戶
    permission_classes = []   #通常主頁就不須要權限驗證了
    throttle_classes = [AnonThrottle,UserThrottle,]  #對匿名用戶和普通用戶的訪問限制

    def get(self,request):
        # self.dispatch
        return Response('訪問首頁')

    def throttled(self, request, wait):
        '''可定製方法設置中文錯誤'''

        # raise exceptions.Throttled(wait)
        class MyThrottle(exceptions.Throttled):
            default_detail = '請求被限制'
            extra_detail_singular = 'Expected available in {wait} second.'
            extra_detail_plural = 'Expected available in {wait} seconds.'
            default_code = '還須要再等{wait}秒'

        raise MyThrottle(wait)

#需登陸就能夠訪問
class ManageView(APIView):
    authentication_classes = [MyAuthentcate, ]  # 認證判斷他是否是匿名用戶
    permission_classes = [MyPermission,]  # 通常主頁就不須要權限驗證了
    throttle_classes = [AnonThrottle, UserThrottle, ]  # 對匿名用戶和普通用戶的訪問限制

    def get(self, request):
        # self.dispatch
        return Response('管理人員訪問頁面')

    def throttled(self, request, wait):
        '''可定製方法設置中文錯誤'''

        # raise exceptions.Throttled(wait)
        class MyThrottle(exceptions.Throttled):
            default_detail = '請求被限制'
            extra_detail_singular = 'Expected available in {wait} second.'
            extra_detail_plural = 'Expected available in {wait} seconds.'
            default_code = '還須要再等{wait}秒'

        raise MyThrottle(wait)

 

若是有某些方法不須要頻次，如xxx函數，則須要在xxx函數中單獨加入一個配置屬性：

1
throttle_classes  =  []

示例 

認證

from django.contrib import admin
from django.urls import path,re_path,include

urlpatterns = [
    re_path("testrestfulframework/",include("testrestfulframework.urls")),
]



#################  testrestfulframework.urls ###############
from django.urls import path,re_path,include
from testrestfulframework import views

urlpatterns = [
    re_path(r"login/$",views.LoginViewSet.as_view()),
    re_path(r"books/$",views.BookViewSet.as_view()),
    re_path(r"books/(?P<pk>\d+)$",views.BookDetailViewSet.as_view()),

]
urls.py 
from django.db import models


class Book(models.Model):
    title=models.CharField(max_length=32)
    price=models.IntegerField()
    pub_date=models.DateField()
    publish=models.ForeignKey("Publish",on_delete=models.CASCADE)
    authors=models.ManyToManyField("Author")
    def __str__(self):
        return self.title

class Publish(models.Model):
    name=models.CharField(max_length=32)
    email=models.EmailField()
    def __str__(self):
        return self.name

class Author(models.Model):
    name=models.CharField(max_length=32)
    age=models.IntegerField()
    def __str__(self):
        return self.name


class User(models.Model):
    username = models.CharField(max_length=16)
    password = models.CharField(max_length=64)

class Token(models.Model):
    token = models.CharField(max_length=128)
    user = models.ForeignKey(to=User,on_delete=models.CASCADE)
models.py 
# -*- coding:utf-8 -*-

from rest_framework.authentication import BaseAuthentication
from rest_framework import exceptions
from testrestfulframework import models


class TokenAuth(BaseAuthentication):
    def authenticate(self, request):
        '''函數名必須叫authenticate'''
        # 驗證條件根據需求設置（此示例爲須要有token值）
        token = request.GET.get('token')
        print(request.data)
        print(request.GET)
        token_obj = models.Token.objects.filter(token=token).first()
        if not token_obj:
            # 若是驗證失敗，須要跑出AuthenticationFailed錯誤
            raise exceptions.AuthenticationFailed("驗證失敗!")
        else:
            user = token_obj.user
            # 若是驗證成功，須要返回一個元組，分別是用戶以及驗證類的實例對象，而後內部會賦值給request.user和request.auth
            return user.username, token_obj
service/auth.py 
# -*- coding:utf-8 -*-


import hashlib, time

def get_random_str(user):
    ctime = str(time.time())
    md5 = hashlib.md5(bytes(user, encoding="utf8"))
    md5.update(bytes(ctime, encoding="utf8"))

    return md5.hexdigest()
service/random_token.py 
# -*- coding:utf-8 -*-
from rest_framework import serializers
from testrestfulframework import models


class BookSerializers(serializers.ModelSerializer):
    # publish = serializers.HyperlinkedIdentityField(
    #     view_name='publish_detail',
    #     lookup_field="publish_id",
    #     lookup_url_kwarg="pk")


    class Meta:
        model = models.Book
        fields = "__all__"
        # depth=1


class PublshSerializers(serializers.ModelSerializer):
    class Meta:
        model = models.Publish
        fields = "__all__"
        # depth = 1

class AuthorSerializers(serializers.ModelSerializer):

    class Meta:
        model = models.Author
        fields = "__all__"
        # depth = 1
serializers.py 
<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>

<form action="/testrestfulframework/login/" method="POST">
    {% csrf_token %}
    <div>用戶名：<input type="text" name="username"></div>
    <div>密碼：<input type="password" name="password"></div>
    <input type="submit" value="登陸">
</form>


</body>
</html>
login.html

認證1 局部認證

from django.shortcuts import HttpResponse
from rest_framework import mixins
from rest_framework import generics
from testrestfulframework import models
from testrestfulframework import serializers
from testrestfulframework.service.auth import TokenAuth
from testrestfulframework.service import random_token
from rest_framework.views import APIView
from django.http import JsonResponse
from django.shortcuts import render


class LoginViewSet(APIView):
    # authentication_classes = [TokenAuth, ]

    def get(self,request):
        return render(request,"login.html")

    def post(self, request, *args, **kwargs):
        res = {"code": 1000, "msg": None}
        try:
            username = request._request.POST.get("username")
            password = request._request.POST.get("password")
            user_obj = models.User.objects.filter(username=username, password=password).first()
            print(username, password, user_obj)
            if not user_obj:
                res["code"] = 1001
                res["msg"] = "用戶名或者密碼錯誤"
            else:
                token = random_token.get_random_str(username)
                models.Token.objects.update_or_create(user=user_obj, defaults={"token": token})
                res["token"] = token

        except Exception as e:
            res["code"] = 1002
            res["msg"] = e
        import json
        print(res)
        # return HttpResponse(json.dumps(res))
        return JsonResponse(res, json_dumps_params={"ensure_ascii": False})




class BookViewSet(generics.ListCreateAPIView):
    authentication_classes = [TokenAuth,]

    queryset = models.Book.objects.all()
    serializer_class = serializers.BookSerializers


class BookDetailViewSet(generics.RetrieveUpdateDestroyAPIView):
    queryset = models.Book.objects.all()
    serializer_class = serializers.BookSerializers
views.py

第一步：登陸

第二步，手動提取token，並以get方式訪問目標內容。

注意：這裏是手動，實際上徹底能夠在views.py經過url處理redirect到目標url，此處不作這操做。

不帶token

 帶token

不作驗證的url可直接訪問

 

 認證2 全局認證

REST_FRAMEWORK = {
    "DEFAULT_AUTHENTICATION_CLASSES": ["testrestfulframework.service.auth.TokenAuth",]
}
settings.py 
from django.shortcuts import HttpResponse
from rest_framework import mixins
from rest_framework import generics
from testrestfulframework import models
from testrestfulframework import serializers
from testrestfulframework.service.auth import TokenAuth
from testrestfulframework.service import random_token
from rest_framework.views import APIView
from django.http import JsonResponse
from django.shortcuts import render


class LoginViewSet(APIView):
    # authentication_classes = [TokenAuth, ]

    def get(self,request):
        return render(request,"login.html")

    def post(self, request, *args, **kwargs):
        res = {"code": 1000, "msg": None}
        try:
            username = request._request.POST.get("username")
            password = request._request.POST.get("password")
            user_obj = models.User.objects.filter(username=username, password=password).first()
            print(username, password, user_obj)
            if not user_obj:
                res["code"] = 1001
                res["msg"] = "用戶名或者密碼錯誤"
            else:
                token = random_token.get_random_str(username)
                models.Token.objects.update_or_create(user=user_obj, defaults={"token": token})
                res["token"] = token

        except Exception as e:
            res["code"] = 1002
            res["msg"] = e
        import json
        print(res)
        # return HttpResponse(json.dumps(res))
        return JsonResponse(res, json_dumps_params={"ensure_ascii": False})




class BookViewSet(generics.ListCreateAPIView):
    # authentication_classes = [TokenAuth,]

    queryset = models.Book.objects.all()
    serializer_class = serializers.BookSerializers


class BookDetailViewSet(generics.RetrieveUpdateDestroyAPIView):
    authentication_classes = []
    
    queryset = models.Book.objects.all()
    serializer_class = serializers.BookSerializers
views.py

 驗證結果同上，略。

權限

url.py、serializers.py、models.py均與上例一致。

from rest_framework.permissions import BasePermission

class SVIPPermission(BasePermission):
    message = '沒有訪問權限,請充值超級會員'

    def has_permission(self, request,view):
        # auth認證裏面找到user對象的type字段值
        role = request.auth.user.role

        if role == 3:
            return True
        else:
            return False


    def has_object_permission(self, request,view,obj):
        # 若是是單個單個條目的detail，須要has_object_permission，此處，obj是該條目
        # 可作進一步處理
        print(obj)  # b001
        print(type(obj))  # <class 'testrestfulframework.models.Book'>
        return True
service/permission.py

權限1 局部權限

from django.shortcuts import HttpResponse
from rest_framework import mixins
from rest_framework import generics
from testrestfulframework import models
from testrestfulframework import serializers
from testrestfulframework.service.auth import TokenAuth
from testrestfulframework.service.permissions import SVIPPermission
from testrestfulframework.service import random_token
from rest_framework.views import APIView
from django.http import JsonResponse
from django.shortcuts import render


class LoginViewSet(APIView):
    # authentication_classes = [TokenAuth, ]

    def get(self,request):
        return render(request,"login.html")

    def post(self, request, *args, **kwargs):
        res = {"code": 1000, "msg": None}
        try:
            username = request._request.POST.get("username")
            password = request._request.POST.get("password")
            user_obj = models.User.objects.filter(username=username, password=password).first()
            print(username, password, user_obj)
            if not user_obj:
                res["code"] = 1001
                res["msg"] = "用戶名或者密碼錯誤"
            else:
                token = random_token.get_random_str(username)
                models.Token.objects.update_or_create(user=user_obj, defaults={"token": token})
                res["token"] = token
                res["user"] = user_obj.username

        except Exception as e:
            res["code"] = 1002
            res["msg"] = e
        import json
        # return HttpResponse(json.dumps(res))
        return JsonResponse(res, json_dumps_params={"ensure_ascii": False})




class BookViewSet(generics.ListCreateAPIView):
    authentication_classes = (TokenAuth,)
    permission_classes = (SVIPPermission,)

    queryset = models.Book.objects.all()
    serializer_class = serializers.BookSerializers


class BookDetailViewSet(generics.RetrieveUpdateDestroyAPIView):
    authentication_classes = (TokenAuth,)
    permission_classes = (SVIPPermission,)

    queryset = models.Book.objects.all()
    serializer_class = serializers.BookSerializers
views.py

權限2 全局權限

REST_FRAMEWORK = {
    # "DEFAULT_AUTHENTICATION_CLASSES": ("testrestfulframework.service.auth.TokenAuth",),
    'DEFAULT_PERMISSION_CLASSES': ('testrestfulframework.service.permissions.SVIPPermission',)
}
settings.py 
from django.shortcuts import HttpResponse
from rest_framework import mixins
from rest_framework import generics
from testrestfulframework import models
from testrestfulframework import serializers
from testrestfulframework.service.auth import TokenAuth
from testrestfulframework.service.permissions import SVIPPermission
from testrestfulframework.service import random_token
from rest_framework.views import APIView
from django.http import JsonResponse
from django.shortcuts import render


class LoginViewSet(APIView):
    # authentication_classes = [TokenAuth, ]

    def get(self,request):
        return render(request,"login.html")

    def post(self, request, *args, **kwargs):
        res = {"code": 1000, "msg": None}
        try:
            username = request._request.POST.get("username")
            password = request._request.POST.get("password")
            user_obj = models.User.objects.filter(username=username, password=password).first()
            print(username, password, user_obj)
            if not user_obj:
                res["code"] = 1001
                res["msg"] = "用戶名或者密碼錯誤"
            else:
                token = random_token.get_random_str(username)
                models.Token.objects.update_or_create(user=user_obj, defaults={"token": token})
                res["token"] = token
                res["user"] = user_obj.username

        except Exception as e:
            res["code"] = 1002
            res["msg"] = e
        import json
        # return HttpResponse(json.dumps(res))
        return JsonResponse(res, json_dumps_params={"ensure_ascii": False})




class BookViewSet(generics.ListCreateAPIView):
    authentication_classes = (TokenAuth,)
    # permission_classes = (SVIPPermission,)

    queryset = models.Book.objects.all()
    serializer_class = serializers.BookSerializers


class BookDetailViewSet(generics.RetrieveUpdateDestroyAPIView):
    authentication_classes = (TokenAuth,)
    # permission_classes = (SVIPPermission,)

    queryset = models.Book.objects.all()
    serializer_class = serializers.BookSerializers
views.py

結果同上，略。

頻率

 頻率1 局部頻率

# -*- coding:utf-8 -*-
from rest_framework.throttling import BaseThrottle

VISIT_RECORD = {}


class VisitThrottle(BaseThrottle):

    def __init__(self):
        self.history = None

    def allow_request(self, request, view):
        remote_addr = request.META.get('REMOTE_ADDR')
        print(remote_addr)
        import time
        ctime = time.time()

        if remote_addr not in VISIT_RECORD:
            VISIT_RECORD[remote_addr] = [ctime, ]  # 這裏只針對源ip作限制，還能夠經過源ip和目標url進行限制，根據本身的設計進行限制。
            return True

        history = VISIT_RECORD.get(remote_addr)
        self.history = history

        while history and history[-1] < ctime - 60:  # 一分鐘以上的訪問，能夠從history去除，爲history騰出空間
            history.pop()

        if len(history) < 3:  # history空間爲3，
            history.insert(0, ctime)
            return True
        else:
            return False

    def wait(self):
        """
        會返回一個這樣的信息：{"detail":"Request was throttled. Expected available in xxx(這裏是60的倒計時) seconds."}
        """
        import time
        ctime = time.time()
        
        return 60 - (ctime - self.history[-1])
service/throttle.py 
from django.shortcuts import HttpResponse
from rest_framework import mixins
from rest_framework import generics
from testrestfulframework import models
from testrestfulframework import serializers
from testrestfulframework.service.auth import TokenAuth
from testrestfulframework.service.permissions import SVIPPermission
from testrestfulframework.service.throttle import VisitThrottle
from testrestfulframework.service import random_token
from rest_framework.views import APIView
from django.http import JsonResponse
from django.shortcuts import render


class LoginViewSet(APIView):
    # authentication_classes = [TokenAuth, ]

    def get(self,request):
        return render(request,"login.html")

    def post(self, request, *args, **kwargs):
        res = {"code": 1000, "msg": None}
        try:
            username = request._request.POST.get("username")
            password = request._request.POST.get("password")
            user_obj = models.User.objects.filter(username=username, password=password).first()
            print(username, password, user_obj)
            if not user_obj:
                res["code"] = 1001
                res["msg"] = "用戶名或者密碼錯誤"
            else:
                token = random_token.get_random_str(username)
                models.Token.objects.update_or_create(user=user_obj, defaults={"token": token})
                res["token"] = token
                res["user"] = user_obj.username

        except Exception as e:
            res["code"] = 1002
            res["msg"] = e
        import json
        # return HttpResponse(json.dumps(res))
        return JsonResponse(res, json_dumps_params={"ensure_ascii": False})




class BookViewSet(generics.ListCreateAPIView):
    # authentication_classes = (TokenAuth,)
    # permission_classes = (SVIPPermission,)
    throttle_classes = [VisitThrottle]

    queryset = models.Book.objects.all()
    serializer_class = serializers.BookSerializers


class BookDetailViewSet(generics.RetrieveUpdateDestroyAPIView):
    # authentication_classes = (TokenAuth,)
    # permission_classes = (SVIPPermission,)
    throttle_classes = [VisitThrottle]

    queryset = models.Book.objects.all()
    serializer_class = serializers.BookSerializers
views.py 
import requests
import gevent
from gevent import monkey;monkey.patch_all()

url = "http://127.0.0.1:8000/testrestfulframework/books/"
url = "http://127.0.0.1:8000/testrestfulframework/books/2"


def func(url):
    res = requests.request(method="GET",url=url)
    print(res.text)
    return res.text


gevent.joinall([gevent.spawn(func,url=url) for i in range(10)])
客戶端

頻率2 全局頻率

REST_FRAMEWORK = {
    # "DEFAULT_AUTHENTICATION_CLASSES": ("testrestfulframework.service.auth.TokenAuth",),
    # 'DEFAULT_PERMISSION_CLASSES': ('testrestfulframework.service.permissions.SVIPPermission',),
    "DEFAULT_THROTTLE_CLASSES":("testrestfulframework.service.throttle.VisitThrottle",)
}
settings 
from django.shortcuts import HttpResponse
from rest_framework import mixins
from rest_framework import generics
from testrestfulframework import models
from testrestfulframework import serializers
from testrestfulframework.service.auth import TokenAuth
from testrestfulframework.service.permissions import SVIPPermission
from testrestfulframework.service.throttle import VisitThrottle
from testrestfulframework.service import random_token
from rest_framework.views import APIView
from django.http import JsonResponse
from django.shortcuts import render


class LoginViewSet(APIView):
    # authentication_classes = [TokenAuth, ]

    def get(self,request):
        return render(request,"login.html")

    def post(self, request, *args, **kwargs):
        res = {"code": 1000, "msg": None}
        try:
            username = request._request.POST.get("username")
            password = request._request.POST.get("password")
            user_obj = models.User.objects.filter(username=username, password=password).first()
            print(username, password, user_obj)
            if not user_obj:
                res["code"] = 1001
                res["msg"] = "用戶名或者密碼錯誤"
            else:
                token = random_token.get_random_str(username)
                models.Token.objects.update_or_create(user=user_obj, defaults={"token": token})
                res["token"] = token
                res["user"] = user_obj.username

        except Exception as e:
            res["code"] = 1002
            res["msg"] = e
        import json
        # return HttpResponse(json.dumps(res))
        return JsonResponse(res, json_dumps_params={"ensure_ascii": False})




class BookViewSet(generics.ListCreateAPIView):
    # authentication_classes = (TokenAuth,)
    # permission_classes = (SVIPPermission,)
    # throttle_classes = [VisitThrottle]

    queryset = models.Book.objects.all()
    serializer_class = serializers.BookSerializers


class BookDetailViewSet(generics.RetrieveUpdateDestroyAPIView):
    # authentication_classes = (TokenAuth,)
    # permission_classes = (SVIPPermission,)
    throttle_classes = []

    queryset = models.Book.objects.all()
    serializer_class = serializers.BookSerializers
views.py

設置了throttle_classes = []的view

 頻率3 內置throttle類

views.py同上

REST_FRAMEWORK = {
    # "DEFAULT_AUTHENTICATION_CLASSES": ("testrestfulframework.service.auth.TokenAuth",),
    # 'DEFAULT_PERMISSION_CLASSES': ('testrestfulframework.service.permissions.SVIPPermission',),
    "DEFAULT_THROTTLE_CLASSES":("testrestfulframework.service.throttle.VisitThrottle",),
    "DEFAULT_THROTTLE_RATES": {
        "visit_rate": "5/m",  # 1分鐘5次，同理，5/s,5/h,5/d，表示秒/時/日
    }
}
settings.py 
from rest_framework.throttling import BaseThrottle,SimpleRateThrottle


class VisitThrottle(SimpleRateThrottle):
    scope = "visit_rate"

    def get_cache_key(self, request, view):
        return self.get_ident(request)
service/throttle.py

 

參考or轉發

http://www.javashuo.com/article/p-bzhgyogi-ee.html

http://www.javashuo.com/article/p-pbbjfaok-ek.html

https://www.cnblogs.com/kermitjam/p/9416097.html

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程