数据库

class UserInfo(models.Model):
    user_type_choices = (
        (1,'普通用户'),
        (2,'VIP'),
        (3,'SVIP')
    )
    username = models.CharField(max_length=32, unique=True)
    password = models.CharField(max_length=64)
    user_type = models.IntegerField(choices=user_type_choices)

设计了各种权限,包括普通用户,VIP和SVIP,当我们需要对于不用的页面设置不同权限的用户访问的时候。\

权限源码实现

局部配置

跟认证同理,我们可以编写我们的权限认证的类

class SvipPermission(object):
    """
    只有SVIP有权限访问
    """
    def has_permission(self, request, view):
        if request.user.user_type !=3 :
            return False
        return True

class VipPermission(object):
    """
    普通用户和VIP有权限访问
    """
    def has_permission(self, request, view):
        if request.user.user_type ==3 :
            return False
        return True

然后通过在我们的views.py视图函数中引入这个类,通过设置permission_classes即可,跟认证的使用一样

全局配置

    def check_permissions(self, request):
        """
        Check if the request should be permitted.
        Raises an appropriate exception if the request is not permitted.
        """
        for permission in self.get_permissions():
            if not permission.has_permission(request, self):
                self.permission_denied(
                    request, message=getattr(permission, 'message', None)
                )

可以看到首先是在self.get_permissions()里面对权限的对象列表进行循环,同时刚刚为什么是重写has_permission这个函数就可以在这里得到结果

点击get_permission可以看到

    def get_permissions(self):
        """
        Instantiates and returns the list of permissions that this view requires.
        """
        return [permission() for permission in self.permission_classes]

返回了permission_classes的实例化对象,真正的permission_classes是下面这个👇
permission_classes = api_settings.DEFAULT_PERMISSION_CLASSES
通过这个我们就可以知道全局配置该如何去配置

REST_FRAMEWORK = {
    ...  # 认证的配置
    'DEFAULT_PERMISSION_CLASSES':['api.utils.permission.SvipPermission'],
}

访问频率的控制

class VisitThrottle(object):
    """
    60s内只能访问3次
    """

    def __init__(self):
        self.history = None

    def allow_request(self, request, view):
        remote_addr = request._request.META.get('REMOTE_ADDR')  # 获取用户IP
        ctime = time.time()
        if remote_addr not in VISIT_RECORD:
            VISIT_RECORD[remote_addr] = [ctime]
            return True
        history = VISIT_RECORD.get(remote_addr)
        self.history = history
        while history and history[-1] < ctime - 60:
            history.pop()
        if len(history) < 3:
            history.insert(0, ctime)
            return True

    def wait(self):
        """
        动态返回还需要等多少秒
        :return:
        """
        ctime = time.time()
        return 60 - (ctime - self.history[-1])

首先获得访问者的IP
原理如下:

  1. 构建一个字典,键值是用户的访问IP
  2. 往字典中填入用户的访问时间,新的访问时间放在最左边
  3. 如果新的访问时间-最后的访问时间大于60s,则把最后一个访问时间弹出,直到最后一个访问时间跟现在的时间差小于60s
  4. 在列表的最左边填入用户的访问时间的值。
  5. wait函数里面可以返回还需要等待的时间

访问频率源码

依旧是通过dispatch进去,进入到self.check_throttles(request)

 def check_throttles(self, request):
        """
        Check if request should be throttled.
        Raises an appropriate exception if the request is throttled.
        """
        throttle_durations = []
        for throttle in self.get_throttles():
            if not throttle.allow_request(request, self):
                throttle_durations.append(throttle.wait())

        if throttle_durations:
            self.throttled(request, max(throttle_durations))

可以看到为什么我们要重写allow_request方法,并且可以知道self.get_throttles()中是一个列表生成式,返回的是频率控制类的对象列表,跟前面实现的步骤是类似的。

局部配置

通过编写我们的访问频率的类,代码就是前面的访问频率控制实现的代码。
然后在我们需要做访问频率控制的类中,配置throttle_classes即可

全局配置

通过在settings.py里面配置

REST_FRAMEWORK = {
    ...,  # 之前的配置
    'DEFAULT_THROTTLE_CLASSES':['api.utils.throttle.VisitThrottle'],
}

如果某个类不想做访问频率的配置,直接配置throttle_classes=[]即可

通过内置类实现访问频率的控制

from rest_framework.throttling import BaseThrottle
BaseThrottle是访问频率的基类

class BaseThrottle:
    """
    Rate throttling of requests.
    """

    def allow_request(self, request, view):
        """
        Return `True` if the request should be allowed, `False` otherwise.
        """
        raise NotImplementedError('.allow_request() must be overridden')

    def get_ident(self, request):  # 获得访问者的IP
        """
        Identify the machine making the request by parsing HTTP_X_FORWARDED_FOR
        if present and number of proxies is > 0. If not use all of
        HTTP_X_FORWARDED_FOR if it is available, if not use REMOTE_ADDR.
        """
        xff = request.META.get('HTTP_X_FORWARDED_FOR')
        remote_addr = request.META.get('REMOTE_ADDR')
        num_proxies = api_settings.NUM_PROXIES

        if num_proxies is not None:
            if num_proxies == 0 or xff is None:
                return remote_addr
            addrs = xff.split(',')
            client_addr = addrs[-min(num_proxies, len(addrs))]
            return client_addr.strip()

        return ''.join(xff.split()) if xff else remote_addr

    def wait(self):
        """
        Optionally, return a recommended number of seconds to wait before
        the next request.
        """
        return None

可以看到访问频率控制触发是通过allow_request来实现的。其中get_ident是用来获得访问者的IP,wait即是提示还需要等待多少秒的信息

但是基类的allow_request是必须重写方法的,我们可以通过SimpleRateThrottle这个类来实现,它已经完成了主逻辑,其中它的init是这么定义的

def __init__(self):
    if not getattr(self, 'rate', None):
        self.rate = self.get_rate()
    self.num_requests, self.duration = self.parse_rate(self.rate)

可以看到需要得到rate的属性值,如果没有就调用get_rate方法

    def get_rate(self):
        """
        Determine the string representation of the allowed request rate.
        """
        if not getattr(self, 'scope', None):
            msg = ("You must set either `.scope` or `.rate` for '%s' throttle" %
                   self.__class__.__name__)
            raise ImproperlyConfigured(msg)

        try:
            return self.THROTTLE_RATES[self.scope]
        except KeyError:
            msg = "No default throttle rate set for '%s' scope" % self.scope
            raise ImproperlyConfigured(msg)

get_rate方法也是先获取scope的属性值,可以看到是从self.THROTTLE_RATES[self.scope]这样一个字典的方式来获取的。

所以我们需要去我们的全局设置里配置字典项,同时在类里面配置scope这个字段,相当于是应用哪个配置文件。

'DEFAULT_THROTTLE_RATES': {
        'Visitor': '3/m',
        'User': '10/m',
    }

这里我配置了访问者的频率是每分钟可以访问3次,登录用户的访问频率是每分钟10次。同时我们在类里面设置scope字段

class VisitThrottle(SimpleRateThrottle):
    scope = 'Visitor'

    def get_cache_key(self, request, view):
        return self.get_ident(request)


class UserThrottle(SimpleRateThrottle):
    scope = 'User'

    def get_cache_key(self, request, view):
        return request.user.username

当返回拿到频率之后,例如是3/m,通过self``.num_requests``, ``self``.duration = ``self``.parse_rate(``self``.rate)进行处理
parse_rate的函数如下

    def parse_rate(self, rate):
        """
        Given the request rate string, return a two tuple of:
        <allowed number of requests>, <period of time in seconds>
        """
        if rate is None:
            return (None, None)
        num, period = rate.split('/')
        num_requests = int(num)
        duration = {'s': 1, 'm': 60, 'h': 3600, 'd': 86400}[period[0]]
        return (num_requests, duration)

最后得到num_request =3,duration=60,最后执行allow_request

    def allow_request(self, request, view):
        """
        Implement the check to see if the request should be throttled.

        On success calls `throttle_success`.
        On failure calls `throttle_failure`.
        """
        if self.rate is None:
            return True

        self.key = self.get_cache_key(request, view)
        if self.key is None:
            return True

        self.history = self.cache.get(self.key, [])
        self.now = self.timer()

        # Drop any requests from the history which have now passed the
        # throttle duration
        while self.history and self.history[-1] <= self.now - self.duration:
            self.history.pop()
        if len(self.history) >= self.num_requests:
            return self.throttle_failure()
        return self.throttle_success()

key是通过get_cache_key函数拿到的,get_cache_key是需要我们进行重写的。这里相当于就是我们之前设置的那个字典,只不过它这里放入了缓存
对于游客,我们可以返回它的IP作为键值,对于已经登录的用户,我们可以返回它的用户名作为键值。

版权声明:本文为原创文章,版权归 heroyf 所有。
本文链接: https://heroyf.club/2019/07/django_learn4/


“苹果是给那些为了爱选择死亡的人的奖励”