drf 访问频率限制

概述频率限制 一个网站的访问频率限制是非常重要的，访问频率限制做的好可以预防爬虫等恶意行为。 使用drf的频率限制对网站接口访问做出限制十分的便捷好用，你只需要直接进行配置即可。 内置限制 局部使用 首先 频率限制

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

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

内置限制局部使用

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

from rest_framework.throttling import UserRateThrottle  # 已登录的from rest_framework.throttling import AnonRateThrottle  # 未登录的from rest_framework.generics import GenericAPIVIEwfrom rest_framework.response import Responsefrom django.contrib import authclass 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次，

判断访问者的IP是否在这个请求IP的字典里保证这个列表里都是最近10秒内的访问的时间。判断当前请求时间和列表里最早的(也就是最后的)请求时间的如果差大于10秒，说明请求以及不是最近10秒内的，删除掉继续判断倒数第二个，直到差值小于10秒判断列表的长度(即访问次数)，是否大于我们设置的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() # d出最后一位        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

总结

以上是内存溢出为你收集整理的drf 访问频率限制全部内容，希望文章能够帮你解决drf 访问频率限制所遇到的程序开发问题。

如果觉得内存溢出网站内容还不错，欢迎将内存溢出网站推荐给程序员好友。