1. 程式人生 > >restframework認證許可權頻率

restframework認證許可權頻率

認證元件

認證元件的流程

--> CBV裡的APIView --> self.dispatch() --> self.initial() 
--> self.perform_authentication() --> request.user
--> self.initialize_request     --> Request類
--> def user(self) 
--> self._authenticate()    
        for authenticator in self.authenticators:
            
try: user_auth_tuple = authenticator.authenticate(self) except exceptions.APIException: self._not_authenticated() raise if user_auth_tuple is not None: self._authenticator = authenticator self.user, self.auth
= user_auth_tuple return self._not_authenticated()
認證流程

app01.utils.py下

class TokenAuth(BaseAuthentication):
    """自己寫的認證類"""

    def authenticate(self, request):
        token = request.GET.get("token")
        token_obj = Token.objects.filter(token=token).first()
        
if not token_obj: # 認證失敗 raise exceptions.AuthenticationFailed("認證失敗!") else: # print("token_obj.user.name", token_obj.user.name) # print("token_obj.token", token_obj.token) return token_obj.user.name, token_obj.token

區域性的配置

views.py下

from app01.utils import * 
class PublishView(APIView):
    # 認證元件
    authentication_classes = [TokenAuth, ]

    def get(self, request):

        print("token_obj.user.name", request.user)  # 在認證元件註冊進去的
        print("token_obj.token", request.auth)
        publish_list = Publish.objects.all()
        ps = PublisherModerSerializers(publish_list, many=True)
        return Response(ps.data)

    def post(self, request):
        ps = PublisherModerSerializers(data=request.data)
        if ps.is_valid():
            ps.save()
            return Response(ps.data)
        else:
            return Response(ps.errors)

全域性的配置

setting.py下

setting.py
    REST_FRAMEWORK = {
        "DEFAULT_AUTHENTICATION_CLASSES": ("app01.utils.TokenAuth", ),

    }

# 某條介面不想走全域性的認證的話,可以在區域性配一條空的認證配置
authentication_classes = []

許可權元件

許可權元件

app01.utils.py下

class SVIPPermission(object):
    """許可權類"""
    message = "只有超級使用者可以訪問"

    def has_permission(self, request, view):
        username = request.user
        user_type = User.objects.filter(name=username).first().user_type
        if user_type == 3:
            # 通過驗證
            return True
        else:
            # 驗證失敗
            return False

區域性的配置

views.py下

from app01.utils import *

class BookViewSet(generics.ListCreateAPIView):
    permission_classes = [SVIPPermission,]
    queryset = Book.objects.all()
    serializer_class = BookSerializers

全域性的配置

setting.py下

REST_FRAMEWORK={
    "DEFAULT_PERMISSION_CLASSES":["app01.utils.SVIPPermission",]
}

頻率元件

app01.utils.py下

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下

from app01.utils import *

class BookViewSet(generics.ListCreateAPIView):
    throttle_classes = [VisitThrottle,]
    queryset = Book.objects.all()
    serializer_class = BookSerializers

setting.py下

REST_FRAMEWORK={
    "DEFAULT_THROTTLE_CLASSES":["app01.utils.VisitThrottle",]
}

內建throttle類

app01.utils.py下

class VisitThrottle(SimpleRateThrottle):

    scope="visit_rate"
    def get_cache_key(self, request, view):

        return self.get_ident(request)

setting.py下

REST_FRAMEWORK={
    "DEFAULT_THROTTLE_CLASSES":["app01.utils.VisitThrottle",],
    "DEFAULT_THROTTLE_RATES":{
    "visit_rate":"5/m",
}
}