drf 访问频率限制

Django 2021-01-01

频率限制

　　 一个网站的访问频率限制是非常重要的，访问频率限制做的好可以预防爬虫等恶意行为。

　　 使用drf的频率限制对网站接口访问做出限制十分的便捷好用，你只需要直接进行配置即可。

内置限制

局部使用

　　 首先我们在视图中进行配置：

from rest_framework.throttling import UserRateThrottle  # 已登录的
from rest_framework.throttling import AnonRateThrottle  # 未登录的
from rest_framework.generics import GenericAPIView
from rest_framework.response import Response
from django.contrib import auth

class ThrottleTestAPI(GenericAPIView):
    throttle_classes = [UserRateThrottle,AnonRateThrottle]

    def get(self,request):
        if request.user.is_authenticated:  # 用户一登陆
            return Response(data="你的请求次数有十次每分钟")
        return Response("你的请求次数只有三次每分钟")
        
    def post(self,request):
        user_obj = auth.authenticate(username="admin",password="admin123")
        auth.login(request,user_obj)
        return Response("登录成功了")

　　 其次，针对已登录的用户和未登录的用户，可以在settings中进行配置限制次数：

REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_RATES': {  # 限制次数 ， 未登录用户一分钟最多三次，登录用户最多一分钟十次
        'anon': '3/m',# 会去配置的 UserRateThrottle 以及 AnonRateThrottle 中找到属性 scope ，scope对应的就是生效的配置
        'user': '10/m'
    }
}

全局使用

　　 如果全局使用，则可以进行如下配置：

REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_CLASSES': (
        'rest_framework.throttling.AnonRateThrottle','rest_framework.throttling.UserRateThrottle'
    ),'DEFAULT_THROTTLE_RATES': {  # 限制次数 ， 未登录用户一分钟最多三次，登录用户最多一分钟十次
        'anon': '3/m',# 会去配置的 UserRateThrottle 以及 AnonRateThrottle 中找到属性 scope ，scope对应的就是生效的配置
        'user': '10/m'
    }
}

　　 如果想针对某一个视图取消全局配置，则将throttle_classes设置为空列表即可：

class ThrottleTestAPI(GenericAPIView):
    throttle_classes = []

    def get(self,request):
        if request.user.is_authenticated:
            return Response(data="你的请求次数有十次每分钟")
        return Response("你的请求次数只有三次每分钟")

    def post(self,user_obj)
        return Response("登录成功了")

自定制限制

频率限制原理

　　 drf中的频率控制基本原理是基于访问次数和时间的，当然我们可以通过自己定义的方法来实现。当我们请求进来，走到我们频率组件的时候，drf内部会有一个字典来记录访问者的IP

　　 以这个访问者的IP为key，value为一个列表，value里面存放访问者每次访问的时间，如下：

　　 { IP1: [第三次访问时间，第二次访问时间，第一次访问时间]，}

　　 把每次访问最新时间放入列表的最前面，记录这样一个数据结构后，通过什么方式限流呢

　　 如果我们设置的是10秒内只能访问5次，

  1. 判断访问者的IP是否在这个请求IP的字典里
  2. 保证这个列表里都是最近10秒内的访问的时间。判断当前请求时间和列表里最早的(也就是最后的)请求时间的如果差大于10秒，说明请求以及不是最近10秒内的，删除掉继续判断倒数第二个，直到差值小于10秒
  3. 判断列表的长度(即访问次数)，是否大于我们设置的5次，如果大于就限流，否则放行，并把时间放入列表的最前面。

自定义限制

　　 使用自定义限制时，需要创建一个类并且重写allow_request()以及wait()方法。

　　 allow_request()有两个额外的参数，分别是二次包装后的request对象，以及实例化过后的视图类本身，当频率限制通过后返回True，否则返回False。

　　 wait()方法是在return False后触发，必须返回一个int类型的值来回复频率限制还有多久取消。

class RequestLimit:
    request_dict = {}
    def __init__(self):
        self.expiration = None
        self.count = 3  # 设定最大访问次数
        self.seconds = 10  # 设定过期时间，秒为单位

    def allow_request(self,request,view):  # 自动调用该方法
        # 拿出IP
        ip = request.META.get("REMOTE_ADDR")
        import time
        current_time = time.time()
        # 如果ip不在字典中，则添加即可，代表当前已访问了一次
        if not ip in self.request_dict:
            self.request_dict[ip] = [current_time]
            return True
        # 如果在，判断长度是否等于设定的最大访问次数
        if len(self.request_dict[ip]) == self.count:
            # 如果等于最大访问次数，则判断最后一位的时间和当前时间相差是否大于指定的过期时间
            if current_time - self.request_dict[ip][-1] > self.seconds:
                # 如果大于，清空写入
                self.request_dict[ip].clear()
                self.request_dict[ip].append(current_time)
                return True
            else:
                # 如果不大于，说明时间还没过，暂时不能访问，设置多少秒后才能访问
                self.expiration = self.request_dict[ip][-1] + self.seconds
                return False
        # 如果在，长度不大于3，则追加当前时间
        self.request_dict[ip].append(current_time)
        return True

    def wait(self):
        import time
        current_time = time.time()
        result = self.expiration - current_time  # 用过期的时间，减去当前的时间
        return result

局部使用

　　 直接进行局部使用即可，不要再到项目全局文件夹下的settings.py中做额外的配置了。

class ThrottleTestAPI(GenericAPIView):
    throttle_classes = [app01.app01_throttle.RequestLimit]  # 直接使用即可

    def get(self,request):
        return Response("get...")

全局使用

　　 全局使用也不用再规定过期时间，直接在settings.py中配置使用即可：

REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_CLASSES':['app01.app01_throttle.RequestLimit',],}

源码分析

　　 上面提到了内置频率的限制原理，其实针对一登陆用户也就是UserRateThrottle来说，它是对用户id进行限制。

　　 而针对未登录用户来说，则是进行ip限制。

　　 注释里已经说的很清楚了，如何区分是ip还是id限制，则是get_cache_key()这个方法。可以看见，UserRateThrottle是id限制。

class UserRateThrottle(SimpleRateThrottle):
    """
    Limits the rate of API calls that may be made by a given user.

    The user id will be used as a unique cache key if the user is
    authenticated.  For anonymous requests,the IP address of the request will
    be used.
    """
    scope = 'user'

    def get_cache_key(self,view):
        if request.user.is_authenticated:
            ident = request.user.pk
        else:
            ident = self.get_ident(request)

        return self.cache_format % {
            'scope': self.scope,'ident': ident
        }

　　 下面我们来看它是如何实现的频率限制，首先我们知道，当进行频率限制时是会对SimpleRateThrottle进行实例化，那么就SimpleRateThrottle这个类中做了什么事情。

class SimpleRateThrottle(BaseThrottle):

    cache = default_cache  # Django缓存
    timer = time.time  # 当前时间
    cache_format = 'throttle_%(scope)s_%(ident)s'  # 格式化后的字符串，提示频率限制还有多久
    scope = None  # UserRateThrottle中有，就是user
    THROTTLE_RATES = api_settings.DEFAULT_THROTTLE_RATES  # 默认的配置信息

    def __init__(self):
        if not getattr(self,'rate',None):  # 找rate，显然没有
            self.rate = self.get_rate()  # 运行这里
        self.num_requests,self.duration = self.parse_rate(self.rate)

　　 接下来就看get_rate()：

    def get_rate(self):

        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)

　　 默认配置信息：

    'DEFAULT_THROTTLE_RATES': {
        'user': None,'anon': None,},

　　 由于我们会进行全局配置，所以会去全局项目文件夹下的settings.py中找：

    'DEFAULT_THROTTLE_RATES': {  # 限制次数 ， 未登录用户一分钟最多三次，登录用户最多一分钟十次
        'anon': '3/m',# 会去配置的 UserRateThrottle 以及 AnonRateThrottle 中找到属性 scope ，scope对应的就是生效的配置
        'user': '10/m'
    }

　　 也就是说，它会返回10/m:

class SimpleRateThrottle(BaseThrottle):

    cache = default_cache  # Django缓存
    timer = time.time  # 当前时间
    cache_format = 'throttle_%(scope)s_%(ident)s'  # 格式化后的字符串，提示频率限制还有多久
    scope = None  # UserRateThrottle中有，就是user
    THROTTLE_RATES = api_settings.DEFAULT_THROTTLE_RATES  # 默认的配置信息

    def __init__(self):
        if not getattr(self,None):  # 找rate，显然没有
            self.rate = self.get_rate()  # 运行这里  返回10/m
        self.num_requests,self.duration = self.parse_rate(self.rate)  接着向下，执行。

　　 接下来看self.parse_rate(self.rate)

    def parse_rate(self,rate):
        if rate is None:
            return (None,None)
        num,period = rate.split('/')  # 进行拆分， num = 10  period=m
        num_requests = int(num)  # 转换数字
        duration = {'s': 1,'m': 60,'h': 3600,'d': 86400}[period[0]]  # 取'm'这个的第0个，就是m
        return (num_requests,duration) 返回元组，(10,60)

　　 实例化至此完成，当有请求来时，将会自动执行allow_request()这个方法，可以在APIView这里面的dispatch()中的initial()找到check_throttles()方法，它会执行allow_request()方法。

    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())

　　 接下来我们来看SimpleRateThrottle中的allow_request()会怎么做：

    def allow_request(self,view):

        if self.rate is None:  # 不执行
            return True

        self.key = self.get_cache_key(request,view)  # 用户id
        if self.key is None:  # 不执行
            return True

        self.history = self.cache.get(self.key,[])  # 缓存中获取，如果获取不到这个用户id，就做一个列表。类似于{pk:[]}
        self.now = self.timer()  # 获取当前时间

        while self.history and self.history[-1] <= self.now - self.duration:  # 如果[]为真并且[]中的最后一位时间小于或等于设定的时间 60
            self.history.pop() # 弹出最后一位
        if len(self.history) >= self.num_requests: # 判断[]的长度是否大于等于设定的长度
            return self.throttle_failure()  # 验证失败
        return self.throttle_success()  # 验证成功

　　 当验证成功后，它会进行在列表中插入时间：

    def throttle_success(self):
        self.history.insert(0,self.now)  # [当前时间]
        self.cache.set(self.key,self.history,self.duration) # 缓存中存放 {pk:[],超时时间}，过了时间自动清除
        return True # 返回True

　　 当验证失败后，接着回来看APIView这里面的dispatch()中的initial()的check_throttles()方法

    def check_throttles(self,self):  # 验证失败
                throttle_durations.append(throttle.wait())  # 执行wait()

　　 好了，源码分析看到这一步就大概差不多了。

　　 其实看了这么多，主要想说的是登录用户的限制是id，未登录用户的限制是ip

相关文章

Python代码大全，海量代码任你下载
注：所有源代码均实测运行过。所有源代码均已上传CSDN，请有...
【Django】REST_Framework框架——视图集ViewSet和ModelViewSet源码解析
继承APIView和ViewSetMixin；作用也与APIView基本类似，提供...
一、Django基础介绍
一、Django介绍Python下有许多款不同的 Web 框架。Django是重...
【云原生 | Docker】部署 Django & Nginx & Gunicorn
本文从nginx快速掌握到使用，gunicorn快速掌握到使用，实现小...
uniapp微信小程序订阅消息发送服务通知--超详细
uniapp微信小程序订阅消息发送服务通知
ORM全集
Django终端打印SQL语句 1 Setting配置: 2 默认python 使用的...