drf 认证、权限、频率三组件

海棠未雨，梨花先雪，一半春休 / 2023-09-07 / 原文

一、三大认证

1、支持序列化的视图类中，比如APIview，源码中view---> self.dispatch--->self.initial---> initial()

 def initial(self, request, *args, **kwargs):
        self.format_kwarg = self.get_format_suffix(**kwargs)

        neg = self.perform_content_negotiation(request)
        request.accepted_renderer, request.accepted_media_type = neg

        version, scheme = self.determine_version(request, *args, **kwargs)
        request.version, request.versioning_scheme = version, scheme

        # Ensure that the incoming request is permitted
        self.perform_authentication(request)
        self.check_permissions(request)
        self.check_throttles(request)

二、认证组件

判断用户是否登录，数据库是否有值

1、需求:

通过认证组件去认证，没有认证通过的用户不让登录。认证方式前端发来的token值与数据库进行对比

2、models

from django.db import models

class User(models.Model):
    username = models.CharField(max_length=32)
    password = models.CharField(max_length=32)
    # 用户类型
    user_type = models.IntegerField(choices=((1, '2B用户'), (2, '普通用户'), (3, '超级用户')))

# 一对一的关系
class UserToken(models.Model):
    token = models.CharField(max_length=64)
    # OneToOneField本质就是ForeignKey+unique
    user = models.OneToOneField(to=User, on_delete=models.CASCADE)
    # user = models.ForeignKey(to=User, unique=True,on_delete=models.CASCADE)

2、写一个认证 auth.py

继承 BaseAuthentication

from .models import UserToken
from rest_framework.authentication import BaseAuthentication
from rest_framework.exceptions import AuthenticationFailed

class LoginAuth(BaseAuthentication):
    # 重写一下authenticate方法
    def authenticate(self, request):
        token = request.query_params.get('token')  # 从url中的参数去拿
        print(token) #4d1455f4-97be-4cf6-88a1-dee6e21433ad

        user_token = UserToken.objects.filter(token=token).first()
        if user_token:
            return user_token.user, token
        else:
            raise AuthenticationFailed('您没有登录！')

补充：

token从请求头中去拿

token = request.META.get('HTTP_TOKEN')

token从请求体中取

token = request.data.get('token')

3、views

class UserView(ViewSet):
    # authentication_classes = []  # 局部解除认证禁用，token没有认证通过就不让登录
    permission_classes = []   # 局部解除权限禁用

    @action(methods=['POST'], detail=False)  # /user/login/    post 请求就会执行
    def login(self, request, *args, **kwargs):
        # 前端传入用户名密码
        username = request.data.get('username')
        password = request.data.get('password')
        user = User.objects.filter(username=username, password=password).first()
        print(user)

        if user:
            # 生成一个随机字符串，返回给前端，并且要把随机字符串存到token表中
            token = str(uuid.uuid4())
            ##### 方式一：麻烦方式
            # user_token=UserToken.objects.filter(user=user).first()
            # if user_token:
            #     user_token.token=token
            #     user_token.save()
            # else:
            #     UserToken.objects.create(user=user,token=token)

            ## 方式二：通过user去UserToken表中查，如果能查到用defaults的更新，如果查不到，就用user和defaults新增一条记录
            # 每次登录会返回一个uuid，会更新uuid
            UserToken.objects.update_or_create(defaults={'token': token}, user=user)
            return Response({'code': 100, 'msg': '登录成功', 'token': token, 'username': user.username})

        else:
            return Response({'code': 101, 'msg': '用户名或密码错误'})

补充：

1、继承 ViewSetMixin,APIView，常见的5个路由自动匹配，其他的通过action装饰器进行指定

4、开启认证

在全局开启认证时需要指定app01下面的auth认证模块

## 全局开启
REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'app01.auth.LoginAuth'
    ],
}

##局部开启
authentication_classes = [authenticate]

##局部禁用
authentication_classes = []

5、拿着数据库的token访问

127.0.0.1:8000/user/login/?token=730c7ebf-4b12-4f55-a337-0b341d333395

三、权限组件

判断登录成功后的用户是否有操作权限

1、permissions 文件

继承： BasePermission

from rest_framework.permissions import BasePermission

# 1 写一个类，继承 BasePermission
# 2 重写 has_permission
# 3 在方法中校验用户是否有权限，如果有，就返回True，如果没有，就返回False

class UserPermission(BasePermission):

    def has_permission(self, request, view):
        # request 当次请求的request，  新的，它是在认证类之后执行的，如果认证通过了request.user 就是当前登录用户
        # 拿到当前登录用户，查看它的类型，确定有没有权限
        if request.user.user_type == 3:
            return True
        else:
            self.message = '您的用户类型是：%s，您没有权限操作' % (request.user.get_user_type_display())
            return False

2、views

from rest_framework.viewsets import ViewSetMixin, ViewSet
from rest_framework.response import Response
from rest_framework.decorators import action
from .models import User, UserToken
import uuid

# class UserView(ViewSetMixin,APIView):
class UserView(ViewSet):
    # authentication_classes = []  # 局部解除认证禁用，token没有认证通过就不让登录
    # permission_classes = []   # 局部解除权限禁用

    @action(methods=['POST'], detail=False)  # /user/login/    post 请求就会执行
    def login(self, request, *args, **kwargs):
        # 前端传入用户名密码
        username = request.data.get('username')
        password = request.data.get('password')
        user = User.objects.filter(username=username, password=password).first()
        print(username)

        if user:
            # 生成一个随机字符串，返回给前端，并且要把随机字符串存到token表中
            token = str(uuid.uuid4())
            ##### 方式一：麻烦方式
            # user_token=UserToken.objects.filter(user=user).first()
            # if user_token:
            #     user_token.token=token
            #     user_token.save()
            # else:
            #     UserToken.objects.create(user=user,token=token)

            ## 方式二：通过user去UserToken表中查，如果能查到用defaults的更新，如果查不到，就用user和defaults新增一条记录
            # 每次登录会返回一个uuid，会更新uuid
            UserToken.objects.update_or_create(defaults={'token': token}, user=user)
            return Response({'code': 100, 'msg': '登录成功', 'token': token, 'username': user.username})

        else:
            return Response({'code': 101, 'msg': '用户名或密码错误'})

3、开启权限认证

## 全局认证
REST_FRAMEWORK = {
    'DEFAULT_PERMISSION_CLASSES': [
        'app01.permissions.UserPermission',
    ],
}

## 局部认证
permission_classes = [UserPermission]

##局部解除权限认证
permission_classes = []

4、访问效果

四、频率组件

限制用户ip频繁访问

1、throtting 文件

继承：SimpleRateThrottle

from rest_framework.throttling import SimpleRateThrottle

class IPRateThrottle(SimpleRateThrottle):
    scope = 'auth1'  # scope： 范围

    def get_cache_key(self, request, view):
        print(request.META)
        return request.META.get('REMOTE_ADDR')  # REMOTE_ADDR 远程的ip地址

补充：

1 重写SimpleRateThrottle中的get_cache_key方法，这个方法返回什么就会用什么去做限制

2 以用户id做限制

    def get_cache_key(self, request, view):  # 重写get_cache_key,返回什么，就以什么做限制： IP地址，用户id限制
        # print(request.META)  # request.META 是一个包含有关HTTP请求的元数据的字典
        # return request.META.get('REMOTE_ADDR')  # REMOTE_ADDR 远程的ip地址
        try:
            if request.user:
                user_id = request.user.pk
                print('用户id是:', user_id)
                return user_id
        except Exception:
            raise AuthenticationFailed('您没有登录，请先登录吧')

3 scope 是一个类属性，它的值用在全局配置中

'DEFAULT_THROTTLE_RATES': {
        'auth1': '3/m',  # 一分钟访问三次
    },

2、views

class UserView(ViewSet):
    authentication_classes = []  # 局部解除认证禁用，token没有认证通过就不让登录
    permission_classes = []   # 局部解除权限禁用

    @action(methods=['POST'], detail=False)  # /user/login/    post 请求就会执行
    def login(self, request, *args, **kwargs):
        # 前端传入用户名密码
        username = request.data.get('username')
        password = request.data.get('password')
        user = User.objects.filter(username=username, password=password).first()
        print(username)

        if user:
            # 生成一个随机字符串，返回给前端，并且要把随机字符串存到token表中
            token = str(uuid.uuid4())
            ##### 方式一：麻烦方式
            # user_token=UserToken.objects.filter(user=user).first()
            # if user_token:
            #     user_token.token=token
            #     user_token.save()
            # else:
            #     UserToken.objects.create(user=user,token=token)

            ## 方式二：通过user去UserToken表中查，如果能查到用defaults的更新，如果查不到，就用user和defaults新增一条记录
            # 每次登录会返回一个uuid，会更新uuid
            UserToken.objects.update_or_create(defaults={'token': token}, user=user)
            return Response({'code': 100, 'msg': '登录成功', 'token': token, 'username': user.username})

        else:
            return Response({'code': 101, 'msg': '用户名或密码错误'})

3、开启频率认证

## 全局认证
REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_RATES': {
        'auth1': '3/m',  # 一分钟访问三次
    },
    'DEFAULT_THROTTLE_CLASSES': ['app01.throtting.IPRateThrottle'],
}

## 局部认证
throttle_classes = [IPRateThrottle] 

##局部解除权限认证
throttle_classes = []

4、效果

五、排序使用

1、导入模块 OrderingFilter

from rest_framework.filters import OrderingFilter

导入模块之后，写一个类属性，指定按哪个字段排序

filter_backends = [OrderingFilter]
ordering_fields = ['id', 'user_type']

2、views

这个案例中，能排序的前提是视图类至少继承 GenericViewSet + ListModelMixin, 和数据库打交道，需要序列化数据返回

解释：GenericViewSet 视图类继承了ViewSetMixin（重写了as_view, 改变了路由的新写法）+ GenericAPIView。ListModelMixin 重写了list（查询所有）方法

class UserView(GenericViewSet, ListModelMixin):
    throttle_classes = []
    authentication_classes = []
    permission_classes = []

    queryset = User.objects.all()
    serializer_class = UserSerializer
    
    ## 排序功能
    filter_backends = [OrderingFilter]
    ordering_fields = ['id', 'user_type']

    @action(methods=['POST'], detail=False)  # /user/login/    post 请求就会执行
    def login(self, request, *args, **kwargs):
        # 前端传入用户名密码
        username = request.data.get('username')
        password = request.data.get('password')
        user = User.objects.filter(username=username, password=password).first()

        if user:
            # 生成一个随机字符串，返回给前端，并且要把随机字符串存到token表中
            # 随机字符串使用uuid生成
            token = str(uuid.uuid4())
            # 把随机字符串存到token表中会有两种情况（如果之前没有登录过就是新增，如果之前登录过修改）
            # 先去UserToken表中，根据user查，如果能查到，就修改，查不到就新增一条记录

            ## 方式二：通过user去UserToken表中查，如果能查到用defaults的更新，如果查不到，就用user和defaults新增一条记录
            UserToken.objects.update_or_create(defaults={'token': token}, user=user)
            return Response({'code': 100, 'msg': '登录成功', 'token': token, 'username': user.username})

        else:
            return Response({'code': 101, 'msg': '用户名或密码错误'})

3、访问方式

get方法查询所有用户，在url地址中params传参数

先按照用户类型升序排，再按照id降序排

127.0.0.1:8000/user/?ordering=user_type,id