今日内容概要

• 权限类使用
• 频率类使用
• 认证类源码分析
• 权限类源码分析
• 部分频率类源码分析
• 鸭子类型

今日内容详细

权限类使用

第一步：写一个类，继承BasePermission
第二步：重写has_permission方法
第三步：在方法中校验用户是否有权限(request.user就是当前登录用户)
第四步：如果有权限，返回True,没有权限，返回False
第五步：self.message 是给前端的提示信息
第六步：局部使用
    class UserView(ModelViewSet):
        permission_classes = [UserTypePermission, ]
全局使用
    REST_FRAMEWORK = {
        'DEFAULT_PERMISSION_CLASSES': ['app01.permission.UserTypePermission'],
    }
局部禁用
    permission_classes = []

代码演示

from rest_framework.permissions import BasePermission

class UserTypePermission(BasePermission):
    def has_permission(self, request, view):   # (权限类的对象,request,视图类的对象)
        # 只有超级管理员有权限
        if request.user.user_type == 1:
            return True  # 有权限
        else:
            # self.message = '普通用户和2b用户都没有权限' # 返回给前端的提示是什么样
            # 使用了choice后，user.user_type 拿到的是数字类型，想变成字符串 user.get_user_type_display()
            self.message = '您是：%s ，您没有权限'%request.user.get_user_type_display()
            return False  # 没有权限

频率类使用

无论是否登录和是否有权限，都要限制访问频率，比如一分钟3次

第一步：写一个类继承SimpleRateThrottle
第二步：重写get_cache__key，返回唯一的字符串，会以这个字符串做频率限制
第三步：写一个类属性scop='随意写'，必须要跟配置文件对应
第四步：配置文件中写
REST_FRAMEWORK = {
'DEFAULT_THROTTLE_RATES': { '随意写': '3/m'  # 3/h  3/s  3/d
                          }
}
第五步：局部配置
    class UserView(ModelViewSet):
        throttle_classes = [MyThrottling,]
全局配置
    REST_FRAMEWORK = {
        'DEFAULT_THROTTLE_CLASSES':['app01.throttling.MyThrottling'],
    }
局部禁用
    throttle_classes = []

代码演示

from rest_framework.throttling import BaseThrottle, SimpleRateThrottle
class MyThrottling(SimpleRateThrottle):  # 我们继承SimpleRateThrottle去写，而不是继承BaseThrottle去写
    # 类属性,这个类属性可以随意命名，但要跟配置文件对应
    scope = 'luffy'
    def get_cache_key(self, request, view):
        # 返回什么，频率就以什么做限制
        # 可以通过用户id限制
        # 可以通过ip地址限制
        return request.META.get('REMOTE_ADDR')


# settings.py
REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_CLASSES':['app01.throttling.MyThrottling'],
    'DEFAULT_THROTTLE_RATES': {
        'luffy': '3/m'
    }
}

认证类源码分析

# 写个认证类，重写某个方法，配置在视图类上，就有认证了---》认证类加了，在视图类的方法中，request.user就是当前登录用户---》猜认证类的执行，是在在视图类的方法之前执行的

# 源码分析：
APIView的执行流程：包装了新的request，执行了3大认证权限频率，执行视图类的方法，处理了全局异常

入口：APIView的dispatch的496行上下：self.initial(request, *args, **kwargs)
# self是谁？self是个对象，是谁的对象？ 是视图类的对象---》BookView这些视图类
APIView的initial方法的413行上下有三行代码，分别是认证，权限，频率
    self.perform_authentication(request)  # 认证
    self.check_permissions(request)  # 权限
    self.check_throttles(request)  # 频率

1.读认证类的源码 APIView的perform_authentication(request)方法，316行上下
    def perform_authentication(self, request):
        request.user  # 新的request,因为继承了APIView
2.request是新的request,来到Request类中找user方法，user是个方法被包装成了属性
user方法在Request类中220行上下
    @property
    def user(self):
        if not hasattr(self, '_user'):  # Request类对象中反射_user
            with wrap_attributeerrors():
                self._authenticate()  # 第一次一定会走这个代码，最开始，是没有_user的，执行这句
        return self._user  # 返回_user
3.Request类中的self._authenticate() 373行上下
    def _authenticate(self):
        for authenticator in self.authenticators:  # 配置在视图类中所有的认证类对象
        # 你配置在视图类上authentication_classes = [你写的认证类,]----》[你写的认证类(),你写的认证类2()]
            try:
              # 调用认证类对象的authenticate方法，传了几个？2个，一个认证类自己，一个是self：Request类的对象
              # user_auth_tuple=（user_token.user, token）
                user_auth_tuple = authenticator.authenticate(self)  # 调用认证类对象的authenticate
            except exceptions.APIException:  # 抛的是AuthenticationFailed，捕获的是APIException
                self._not_authenticated()
                raise

            if user_auth_tuple is not None:
                self._authenticator = authenticator  # 忽略
                self.user, self.auth = user_auth_tuple  # 解压赋值 self.user=user_token.user，self.auth=token
                return
                # 认证类可以配置多个，但是如果有一个返回了两个值，后续的就不执行了
        self._not_authenticated()
4. self.authenticators到底是啥？ 
你配置在视图类上authentication_classes = [你写的认证类,]----》[你写的认证类(),你写的认证类2()]
Reqeust这个类实例化的时候，传入的，如果不传就是空元组
找Request的实例化---》dispatch中包装了新的Request

APIView中的dispatch方法：492行上下
    request = self.initialize_request(request, *args, **kwargs)
APIView中的initialize_request方法：394行上下
    authenticators=self.get_authenticators(),

APIView中的get_authenticators方法：268行上下
    def get_authenticators(self):
        # self.authentication_classes配在视图类上的认证类列表---》认证类对象列表
        return [auth() for auth in self.authentication_classes]

总结：认证类需要重写authenticate方法，认证通过返回两个值或None，认证不通过抛AuthenticationFailed（继承了APIException）异常。

权限类源码分析

入口：APIView的dispatch的496行上下：self.initial(request, *args, **kwargs)
# self是谁？self是个对象，是谁的对象？ 是视图类的对象---》BookView这些视图类
APIView的initial方法的413行上下有三行代码，分别是认证，权限，频率
    self.perform_authentication(request)  # 认证
    self.check_permissions(request)  # 权限
    self.check_throttles(request)  # 频率

1.APIView的check_permissions(request)方法：326行上下
    def check_permissions(self, request):
        for permission in self.get_permissions():
            # permission是咱们配置在视图类中权限类的对象，对象调用它的绑定方法has_permission，这就是为什么要重写has_permission方法的原因
            # 对象调用自己的绑定方法会把自己传入(权限类的对象,request,视图类的对象)
            if not permission.has_permission(request, self):
                self.permission_denied(
                    request,
                    message=getattr(permission, 'message', None), # 错误提示信息
                    code=getattr(permission, 'code', None)  # 状态码
                )
2.APIVIew的self.get_permissions()方法：274行上下
    def get_permissions(self):
        return [permission() for permission in self.permission_classes]
self.permission_classes 就是咱们在视图类中配的权限类的列表，所以这个get_permissions返回的是 咱们在视图类中配的权限类的对象列表[UserTypePermession(),]。

总结：权限类要重写has_permission方法，有三个参数，为什么一定要return True或False，messgage可以做什么用。

部分频率类源码分析

入口：APIView的dispatch的496行上下：self.initial(request, *args, **kwargs)
# self是谁？self是个对象，是谁的对象？ 是视图类的对象---》BookView这些视图类
APIView的initial方法的413行上下有三行代码，分别是认证，权限，频率
    self.perform_authentication(request)  # 认证
    self.check_permissions(request)  # 权限
    self.check_throttles(request)  # 频率

APIView的check_throttles方法：352行上下
    def check_throttles(self, request):
        throttle_durations = []
        for throttle in self.get_throttles():  # 配在视图类上频率类列表 频率类的对象
            if not throttle.allow_request(request, self):
                throttle_durations.append(throttle.wait())

        if throttle_durations:
            # Filter out `None` values which may happen in case of config / rate
            # changes, see #1438
            durations = [
                duration for duration in throttle_durations
                if duration is not None
            ]

            duration = max(durations, default=None)
            self.throttled(request, duration)

总结：要写频率类，必须重写allow_request方法，返回True就是没有被频率限制住，返回False就是被频率限制了。

鸭子类型

指的是在面向对象中，子类不需要显示继承的某个类，只要有某个类的方法和属性，那我就属于这个类。

# 假设有个鸭子类Duck类，有两个方法，run，speak方法
# 假设又有一个普通鸭子类,PDuck，如果它也是鸭子，它需要继承Duck类，只要继承了鸭子类，什么都不需要写，普通鸭子类的对象就是鸭子这种类型；如果不继承，普通鸭子类的对象就不是鸭子这种类型
#假设又有一个唐老鸭子类,TDuck，如果它也是鸭子，它需要继承Duck类，只要继承了鸭子类，什么都不需要写，唐老鸭子类的对象就是鸭子这种类型；如果不继承，唐老鸭子类的对象就不是鸭子这种类型

# python不推崇这个，它推崇鸭子类型，指的是
不需要显示的继承某个类，只要我的类中有run和speak方法，我就是鸭子这个类

# 有小问题：如果使用python鸭子类型的写法，如果方法写错了，它就不是这个类型了，会有问题
# python为了解决这个问题：
方式一：abc模块，装饰后，必须重写方法，不重写就报错
方式二：drf源码中使用的：父类中写这个方法，但没有具体实现，直接抛异常

注意

django的配置文件不要乱导入，乱导入可能会出错

因为django的运行是在加载完配置文件后才能运行，模块的导入会执行那个模块，而这个模块中又有别的导入，别的导入又必须在django运行起来之后才能使用

作业

# 1 编写图书和出版社的5个接口，所有接口都要有一分钟访问5次的频率限制
# 2 图书的接口需要登录才能方法，出版社的接口需要登录，并且是超级用户才能访问
# 3 继承BaseThrottle编写频率类
    #class OurThrottling(BaseThrottle):
        #def allow_request(self, request, view):
            # 自定义的逻辑  {ip:[时间1，时间2]}
            # （1）取出访问者ip
            # （2）判断当前ip不在访问字典里，添加进去，并且直接返回True,表示第一次访问，在字典里，继续往下走
            # （3）循环判断当前ip的列表，有值，并且当前时间减去列表的最后一个时间大于60s，把这种数据pop掉，这样列表中只有60s以内的访问时间，
            # （4）判断，当列表小于3，说明一分钟以内访问不足三次，把当前时间插入到列表第一个位置，返回True，顺利通过
            # （5）当大于等于3，说明一分钟内访问超过三次，返回False验证失败
            #return False


# views.py
from django.shortcuts import render

# Create your views here.
from rest_framework.viewsets import ModelViewSet, ViewSet
from .models import Book, Publish, User, UserToken
from .serializer import BookSerializer, PublishSerializer
from rest_framework.decorators import action
import uuid
from rest_framework.response import Response
from .auth import UserThrottling


class BookView(ModelViewSet):
    throttle_classes = [UserThrottling, ]
    permission_classes = []
    queryset = Book.objects.all()
    serializer_class = BookSerializer


class PublishView(ModelViewSet):
    throttle_classes = [UserThrottling, ]
    queryset = Publish.objects.all()
    serializer_class = PublishSerializer


class UserView(ViewSet):
    throttle_classes = [UserThrottling, ]
    authentication_classes = []
    permission_classes = []

    @action(methods=['POST', ], detail=False, url_path='login')
    def login(self, request):
        username = request.data.get('username')
        password = request.data.get('password')
        user = User.objects.filter(username=username, password=password).first()
        if user:
            token = str(uuid.uuid4())
            UserToken.objects.update_or_create(defaults={'token': token}, user=user)
            return Response({'code': 100, 'msg': '登录成功', 'token': token})
        return Response({'code': 101, 'msg': '用户名或密码错误'})


# auth.py
from .models import UserToken
from rest_framework.authentication import BaseAuthentication
from rest_framework.exceptions import AuthenticationFailed
from rest_framework.permissions import BasePermission
from rest_framework.throttling import SimpleRateThrottle, BaseThrottle
import time


class LoginAuth(BaseAuthentication):
    def authenticate(self, request):
        token = request.GET.get('token')
        user_token = UserToken.objects.filter(token=token).first()
        if user_token:
            return user_token.user, token
        raise AuthenticationFailed('您没有登录')


class UserPermission(BasePermission):
    def has_permission(self, request, view):
        if request.user.user_type == 1:
            return True
        return False


# class UserThrottling(SimpleRateThrottle):
#     scope = 'frequency'
#
#     def get_cache_key(self, request, view):
#         return request.META.get('REMOTE_ADDR')


ipdict = {}


class UserThrottling(BaseThrottle):
    def allow_request(self, request, view):
        userip = request.META.get('REMOTE_ADDR')
        ctime = time.time()
        if userip not in ipdict:
            ipdict[userip] = [ctime, ]
            return True
        timelis = ipdict.get(userip)
        while timelis and timelis[-1] < ctime - 60:
            timelis.pop()
        if len(timelis) < 5:
            timelis.insert(0, ctime)
            return True
        else:
            return False


# settings.py
REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': ['app01.auth.LoginAuth', ],
    'DEFAULT_PERMISSION_CLASSES': ['app01.auth.UserPermission', ],
    'DEFAULT_THROTTLE_CLASSES': ['app01.auth.UserThrottling'],
    'DEFAULT_THROTTLE_RATES': {'frequency': '5/m'
                               }
}


# urls.py
from django.contrib import admin
from django.urls import path, include
from rest_framework.routers import SimpleRouter
from app01 import views

router = SimpleRouter()

router.register('user', views.UserView, 'user')
router.register('books', views.BookView)
router.register('publish', views.PublishView)

urlpatterns = [
    path('admin/', admin.site.urls),
    path('', include(router.urls)),
]