rest_framework组件 认证、权限、频率

rest_framework组件

 

认证源码流程图：

 
APIView类
第一步class OrderView(APIView):点击APIView
第二步def dispatch(self, request, *args, **kwargs):
找到self.initial(request, *args, **kwargs)，点击initial
第三步def initial(self, request, *args, **kwargs):
找到self.perform_authentication(request)（进行认证)
点击perform_authentication
第四步def perform_authentication(self, request):
request.user

Request类
第五步同上def dispatch(self, request, *args, **kwargs):
找到request = self.initialize_request(request, *args, **kwargs)
点击initialize_request
第六步def initialize_request(self, request, *args, **kwargs):
找到return Request,点击Request
第七步class Request(object):
找到    @property    def user(self):（获取认证对象，进行一步步认证）
找到return self._user   
第八步还是@property    def user(self):里面
找到self._authenticate()点击_authenticate
第九步def _authenticate(self): （循环所有对象，执行认证方法）

Authtication自定义认证类(自己写了认证方法，就用自己的)
第十步def authenticate(self, request):
报错:(返回元祖(user对象,token对象))

class OrderView(APIView):或class AuthView(APIView):等订单
第十一步执行各个订单的def post:  或def get:等方法

认证组件

局部认证

在需要认证的视图类里加上authentication_classes = [认证组件1类名,认证组件2类名....]

示例如下：

seralizers.py

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.py

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

views.py

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里配置如下：

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

　　这样配置之后，每个视图类都要经过认证成功之后才能执行下一步，

如果有某些方法不需要认证，如login函数，则需要在login函数中单独加入一个配置属性：

1
authentication_classes = [] #自己的类里有的话就调用此类的配置，为空既什么都不做

权限认证

局部认证

permission.py

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

频率认证

 

局部认证

格式如下：

1
2
3
4
5
6
7
8
9
class VisitRateThrottle(object):
    def allow_request(self,request,view):
        # 要求访问站点的频率不能超过每分钟20次等等
        if 1:
            # 如果在限制之内，则返回True
 
            return True
        else:
            return False

　　然后在需要进行限制的视图类中加入如下配置：

1
throttle_classes = [VisitRateThrottle]

全局认证

一  自定义频率控制类

 
class MyThrottle():
    visitor_dic = {}

    def __init__(self):
        self.history = None

    def allow_request(self, request, view):
        '''
            {'ip1':[时间1 ,时间2],
            'ip2':[时间1, ],
            }
           #（1）取出访问者ip
            # （2）判断当前ip不在访问字典里，添加进去，并且直接返回True,表示第一次访问，在字典里，继续往下走
            # （3）循环判断当前ip的列表，有值，并且当前时间减去列表的最后一个时间大于60s，把这种数据pop掉，这样列表中只有60s以内的访问时间，
            # （4）判断，当列表小于3，说明一分钟以内访问不足三次，把当前时间插入到列表第一个位置，返回True，顺利通过
            # （5）当大于等于3，说明一分钟内访问超过三次，返回False验证失败
        '''

        # Meta:请求所有的东西的字典
        # 拿出ip地址
        ip = request.META.get('REMOTE_ADDR')
        # 不在字典中,说明是第一次访问
        ctime = time.time()
        if ip not in self.visitor_dic:
            self.visitor_dic[ip] = [ctime, ]
            return True
        # 根据当前访问者ip,取出访问的时间列表
        history = self.visitor_dic[ip]
        self.history = history
        while history and ctime - history[-1] > 60:
            history.pop()

        if len(history) < 3:
            # 把当前时间放到第0个位置上
            history.insert(0, ctime)
            return True

        return False

    def wait(self):
        # 剩余时间
        ctime = time.time()
        return 60 - (ctime - self.history[-1])
 

二  内置频率控制

在app中新建一个文件,来放相关组件:

from rest_framework.throttling import SimpleRateThrottle
class VisitThrottle(SimpleRateThrottle):
    scope = 'hhh'
    def get_cache_key(self, request, view):
        return self.get_ident(request)

在settings中配置访问:

REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_RATES':{
        'hhh':'3/m'
    }
}

在视图函数中:(局部配置)

throttle_classes = [MyThrottles,]

错误信息提示转换为中文:

 
class Course(APIView):
    authentication_classes = [TokenAuth, ]
    permission_classes = [UserPermission, ]
    throttle_classes = [MyThrottles,]

    def get(self, request):
        return HttpResponse('get')

    def post(self, request):
        return HttpResponse('post')
    def throttled(self, request, wait):
        from rest_framework.exceptions import Throttled
        class MyThrottled(Throttled):
            default_detail = '傻逼啊'
            extra_detail_singular = '还有 {wait} second.'
            extra_detail_plural = '出了 {wait} seconds.'
        raise MyThrottled(wait)
 

其他

内置频率限制类：

BaseThrottle是所有类的基类：方法：def get_ident(self, request)获取标识，其实就是获取ip，自定义的需要继承它

AnonRateThrottle：未登录用户ip限制，需要配合auth模块用

SimpleRateThrottle：重写此方法，可以实现频率现在，不需要咱们手写上面自定义的逻辑

UserRateThrottle:登录用户频率限制，这个得配合auth模块来用

ScopedRateThrottle：应用在局部视图上的（忽略）

内置频率全局配置:

REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_CLASSES':['app01.utils.VisitThrottle',],
    'DEFAULT_THROTTLE_RATES':{
        'hhh':'3/m'
    }
}
【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/xyhh/p/10860694.html