drf权限，频率，过滤，排序，异常处理

目录

• 一、权限
  • 1 权限源码分析
  • 2 自定义权限类
  • 3 内置权限类
• 二、频率
  • 1 内置频率设置
  • 2 自定义频率限制：ip限制
  • 3 内置频率自定义：ip限制
• 三、过滤
• 四、排序
• 五、异常处理

一、权限

1 权限源码分析

# APIView---->dispatch----> self.initial(request, *args, **kwargs)
# ---->self.check_permissions(request)---->
 def check_permissions(self, request):
        # get_permissions是apiview的一个方法，得到的是一个列表
        # 这个列表的生成和认证一模一样，列表生成式内的可迭代对象是在apisettings里配置的
        # 这里我们就可以知道，权限的全局配置，局部配置应该和认证是一模一样的
        # 所以这里循环的列表是一个个权限校验类实例化得到的对象
        for permission in self.get_permissions():
            # 权限校验对象的核心是has_permission方法，返回值必须是bool
            # 这个方法接受了三个参数，权限对象，request，视图类对象
            if not permission.has_permission(request, self):
                self.permission_denied(
                    request, message=getattr(permission, 'message', None)
                )

2 自定义权限类

# uitl
class UserPermission(BasePermission):
    def has_permission(self,request,view):
        # 权限需要根据request.user判断，所以通常会和认证一起使用
        if request.user.user_type == 1:
            return True
        return False
    
# view
class StudentAPI(ModelViewSet):
    authentication_classes = [uitl.TokenAuthentication]
    permission_classes = [uitl.UserPermission]
    queryset = models.Student.objects
    serializer_class = StudentModelSerializer
    
# 局部使用
class TestView(APIView):
    permission_classes = [app_auth.UserPermission]
# 全局使用
REST_FRAMEWORK={
    "DEFAULT_AUTHENTICATION_CLASSES":["app01.app_auth.MyAuthentication",],
    'DEFAULT_PERMISSION_CLASSES': [
        'app01.app_auth.UserPermission',
    ],
}
# 局部禁用
class TestView(APIView):
    permission_classes = []

3 内置权限类

from rest_framework.permissions import IsAdminUser
from rest_framework.authentication import SessionAuthentication
class TestView3(APIView):
    # 如果要用内置的，就要用全套的内置，不容易出错
    authentication_classes=[SessionAuthentication,]
    # 判断用户是否是职员，is_stuff
    permission_classes = [IsAdminUser]
    def get(self,request,*args,**kwargs):
        return Response('这是22222222测试数据，超级管理员可以看')

二、频率

1 内置频率设置

# 全局设置
REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_CLASSES': (
        # 未登陆
        'rest_framework.throttling.AnonRateThrottle',
        # 已登录，由于是内置的，所以若要使用权限和认证也要使用内置的
        'rest_framework.throttling.UserRateThrottle'
    ),
    'DEFAULT_THROTTLE_RATES': {
        # 已登录 一分钟10次
        'user': '10/m',
        # 未登录 一分钟5次
        'anon': '5/m',
    }
}

# 局部配置
class StudentAPI(ModelViewSet):
    throttle_classes = [AnonRateThrottle]
    queryset = models.Student.objects
    serializer_class = StudentModelSerializer
# 这里只设置的频率类，没设置参数，参数只能在全局配置
REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_RATES': {
        # 未登录 一分钟5次
        'anon': '5/m',
    }
}

2 自定义频率限制：ip限制

# 自定义频率检测类
# 频率验证全都会走频率类的all_request方法，如果频率不通过走wait方法，重写这个两个方法即可
# 先取出访问者的ip，判断在不在全局的一个字典中，如果不在，添加这个ip：[time,]（这个列表先进先出）
# 判断这个ip对应的key的长度是否等于3
# 如果等于三，拿出列表[2]判断当时的time-现在的time是否小于60
# 如果大于60，返回false
# 如果小于60，删除最后一个，把当前的请求时间新增进去，返回True
import time

from rest_framework.response import Response


class IP_Throttles():
    visitor_history = {}
    space_time = None
    def allow_request(self,request,view):
        ip = request.META.get('REMOTE_ADDR')
        # 判断是否存在ip访问记录
        if ip in self.visitor_history.keys():
            ip_len = len(self.visitor_history[ip])
            # 判断访问次数是否等于3
            now_time = time.time()
            if ip_len == 3:

                space_time = now_time-self.visitor_history[ip][2]
                # 判断最长间隔是否小于60
                if space_time>60:
                    # 大于60，添加进新访问记录，删除最后一个访问记录
                    self.visitor_history[ip].pop()
                    self.visitor_history[ip].insert(0,now_time)
                else:
                    self.space_time = space_time
                    return False
            else:
                # 如果小于3 添加新访问记录
                self.visitor_history[ip].insert(0,now_time)
        else:
            # 如果没有ip记录，添加新记录
            self.visitor_history[ip] = [time.time()]
        print(self.visitor_history)
        return True
    def wait(self):
        return 60-self.space_time

3 内置频率自定义：ip限制

# 写一个类，继承SimpleRateThrottle，只需要重写get_cache_key 
from rest_framework.throttling import ScopedRateThrottle,SimpleRateThrottle

#继承SimpleRateThrottle
class MyThrottle(SimpleRateThrottle):
    scope='luffy'
    def get_cache_key(self, request, view):
        print(request.META.get('REMOTE_ADDR'))
        return request.META.get('REMOTE_ADDR')   # 返回
    
# 局部使用，全局使用 
REST_FRAMEWORK={
    'DEFAULT_THROTTLE_CLASSES': (
        'utils.throttling.MyThrottle',
    ),
    'DEFAULT_THROTTLE_RATES': {
        'luffy': '3/m'  # key要跟类中的scop对应
    },
}

# python3 manage.py runserver 0.0.0.0:8000   你们局域网就可以相互访问


# 内网穿透

三、过滤

# 安装django-filters
# 在settings注册django-filters
# 在settings配置
REST_FRAMEWORK = {
   'DEFAULT_FILTER_BACKENDS': ('django_filters.rest_framework.DjangoFilterBackend',)

}
# 在视图类中定义可以过滤的字段
class StudentAPI(ModelViewSet):
	# 这里定义了name
    # 那么就可以在url后面用❓拼接筛选条件
    # http://127.0.0.1:8000/student/?nanme=hz&id>1
    filter_fields = ('name','id')
    filter_backends =[DjangoFilterBackend]
    queryset = models.Student.objects
    serializer_class = StudentModelSerializer

四、排序

from rest_framework.filters import OrderingFilter
class StudentAPI(ModelViewSet):	
    # 局部配置内置的排序类
    filter_backends = [OrderingFilter]
    # 限定可排序字段
	ordering_fields = ('id', 'price')
    
    # 按照id倒序，符号为倒序
    http://127.0.0.1:8000/student/?ordering=-id

总结：排序和过滤在局部配置和全局配置都放在一个列表里，只是配置的时候都要加一个限定字段列表，两者可以连用

重点：过滤和排序必须要继承ListAPIView或者与它同类继承的视图类

五、异常处理

# 在drf的settings中
'EXCEPTION_HANDLER': 'rest_framework.views.exception_handler',
# 这表示drf的异常捕获配置方法
def exception_handler(exc, context):
    if isinstance(exc, Http404):
        exc = exceptions.NotFound()
    elif isinstance(exc, PermissionDenied):
        exc = exceptions.PermissionDenied()
    if isinstance(exc, exceptions.APIException):
        headers = {}
        if getattr(exc, 'auth_header', None):
            headers['WWW-Authenticate'] = exc.auth_header
        if getattr(exc, 'wait', None):
            headers['Retry-After'] = '%d' % exc.wait
        if isinstance(exc.detail, (list, dict)):
            data = exc.detail
        else:
            data = {'detail': exc.detail}
        set_rollback()
        return Response(data, status=exc.status_code, headers=headers)
    return None
# 从上面的源码我们可以读取到几个信息
# 1 drf并没有把所有的异常捕获完，因为最后return None表示没有被drf捕获的异常会被django来捕获
# 2 异常是通过apisettings来配置的，也就是我们可以根据这个方法重写捕获异常

# 全局捕获异常
# 现在项目settings中配置

# 我们要在drf捕获异常的基础上再添加捕获，捕获应当有固定的返回信息
from rest_framework.response import Response
from rest_framework.views import exception_handler
def catch_all_exe(exc, context):
    respone = exception_handler(exc, context)
    if not respone:
        return Response(data={'code':404,'msg':f'drf捕获的异常{str(exc)}'},status=404)
    return Response(data={'code':405,'msg':f'django捕获的异常{str(exc)}'},status=405)