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 VisitThrottleBetter(BaseThrottle):
    """优化过代码的频率限制类"""

    def __init__(self):
        self.visited_record = None

    def allow_request(self, request, view):
        """在allow_request中编写频率业务逻辑"""
        remote_addr = request.META.get('REMOTE_ADDR')
        # print(remote_addr)
        ctime = time.time()

        # 1 有记录来则将其添加到访问记录里
        VISIT_RECORD.setdefault(remote_addr, []).insert(0, ctime)

        # 2 判断列表的最后一个是否小于(当前时间-频率时间);如果小于,则剔除;
        self.visited_record = VISIT_RECORD.get(remote_addr)
        while self.visited_record and self.visited_record[-1] < ctime - 60:
            self.visited_record.pop()

        # 3 然后再判断列表的长度是否超过频率数,如不超过,则return true,else false
        if len(self.visited_record) <= 3:
            # /不超过频率数
            return True
        else:
            # /超过频率数
            return False

    def wait(self):
        """如果allow_request中没有通过, wait里写再等多少秒就可以访问了"""
        ctime = time.time()
        first_visit_duration = ctime - self.visited_record[-1]      # 现在离第一次访问过了多久啦
        return 60 - first_visit_duration

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",
}
}