zoukankan      html  css  js  c++  java
  • (三) rest_framework 权限与限流源码梳理

    权限

    rest_framework 执行dispatch-> self.check_permissions(request)方法
    从settings文件获取指定的权限类列表,然后实例化对象。
    settings中指定:DEFAULT_PERMISSION_CLASSES:[]
    然后调用类的has_permission方法,对权限进行验证

    自定义权限认证:

    class Mypermission(BasePermission):
        message = '错误'
        def has_permission(self, request, view):
            token = request.GET.get("token")
            if token == '1':
                return True
            elif token == '2':
                view.permission_denied(
                    request, message=getattr(Mypermission, 'message', None)
                )
            else:
                return False
    

    内置的权限认证类:

    AllowAny , IsAuthenticated(django内置的权限判断is_authenticated方法判断) IsAdminUser IsAuthenticatedOrReadOnly等

    限流

    同样进入视图函数前,initial方法:
    执行self.check_throttles(request)
    settings文件全局添加: DEFAULT_THROTTLE_CLASSES

    限流判断,执行对象的allow_request方法,当返回False时候,执行wait()方法,判断多少秒后可访问。

    自定义限流方法:
    例如每个用户 10s 不能超过5次:

    
    user_dict = {}
    
    class BankAuth(BaseAuthentication):
        def authenticate(self, request):
            user = request._request.GET.get("user", None)
            if not user:
                raise AuthenticationFailed
            return (user, None)
    
        def authenticate_header(self, request):
            pass
    
    class Mythrottle(BaseThrottle):
        def __init__(self):
            self.history = None
    
        def allow_request(self, request, view):
            user = request.user
            if user not in user_dict:
                user_dict[user] = dict(name=user, view_count=1, last_time=time())
                return True
            else:
                second = int(time() - user_dict[user]['last_time'])
                if second < 10:
                    if user_dict[user]['view_count'] > 5:
                        self.history = user_dict[user]['last_time']
                        return False
                else:
                    user_dict[user]['view_count'] = 0
                    user_dict[user]['last_time'] = time()
                user_dict[user]['view_count'] += 1
                return True
                
        def wait(self):
            return 10 - (time() - self.last_time)
    
    class BankView(APIView):
        authentication_classes = [BankAuth, ]
        throttle_classes = [Mythrottle, ]
    
        def get(self, request):
            return JsonResponse(user_dict[request.user])
    

    内置的SimpleRateThrottle方法

    # 首先执行allow_request方法
    # 内部调用get_cache_key获取唯一标识别。待我们自己实现
    #    'DEFAULT_THROTTLE_RATES': {
    #        'my_throttle': '5/m' # ('s', 'sec', 'm', 'min', 'h', 'hour', 'd', #'day')
    #    }
    class Mythrottle(SimpleRateThrottle):
        scope = 'my_throttle' #settings文件中 DEFAULT_THROTTLE_RATES 配置的访问次数限制
        def get_cache_key(self, request, view):
            ident = self.get_ident(request) # 调用父类的获取唯一标识ip 方法
            print(ident)
            return ident
    
  • 相关阅读:
    2021/3/12
    2021/3/11
    2021/3/10
    2021/3/9
    2021/3/8
    2021/3/5
    2021/3/4
    2021/3/3
    2021/3/2
    2021/3/1
  • 原文地址:https://www.cnblogs.com/donghaoblogs/p/12442885.html
Copyright © 2011-2022 走看看