zoukankan      html  css  js  c++  java
  • REST-framework之频率组件

    REST-framework之频率控制

    一 频率简介

    为了控制用户对某个url请求的频率,比如,一分钟以内,只能访问三次

    二 自定义频率类,自定义频率规则

    自定义的逻辑

    """
    1. 取出访问者ip
    2. 判断当前ip不在访问字典里,添加进去,并且直接返回True,表示第一次访问,在字典里,继续往下走
    3. 循环判断当前ip的列表,有值,并且当前时间减去列表的最后一个时间大于60s,把这种数据pop掉,这样列表中只有60s以内的访问时间,
    4. 判断,当列表小于3,说明一分钟以内访问不足三次,把当前时间插入到列表第一个位置,返回True,顺利通过
    5. 当大于等于3,说明一分钟内访问超过三次,返回False验证失败
    """
    

    代码实现:

    class MyThrottles():
        VISIT_RECORD = {}
        def __init__(self):
            self.history=None
    
        def allow_request(self,request, view):
            # 1. 取出访问者ip
            # print(request.META)
            ip=request.META.get('REMOTE_ADDR')
            import time
            ctime=time.time()
            # 2. 判断当前ip不在访问字典里,添加进去,
            # 并且直接返回True,表示第一次访问
            if ip not in self.VISIT_RECORD:
                self.VISIT_RECORD[ip]=[ctime,]
                return True
            self.history=self.VISIT_RECORD.get(ip)
            # 3. 循环判断当前ip的列表,有值,并且当前时间减去列表的最后一个
            # 时间大于60s,把这种数据pop掉,这样列表中只有60s以内的访问时间,
            while self.history and ctime-self.history[-1]>60:
                self.history.pop()
            # 4. 判断,当列表小于3,说明一分钟以内访问不足三次,
            # 把当前时间插入到列表第一个位置,返回True,顺利通过
            # 5. 当大于等于3,说明一分钟内访问超过三次,返回False验证失败
            if len(self.history) < 3 :
                self.history.insert(0,ctime)
                return True
            else:
                return False
    
        def wait(self):
            import time
            ctime=time.time()
            return 60-(ctime-self.history[-1])
    
    

    三 内置频率类及局部使用

    写一个类,继承自SimpleRateThrottle,(根据ip限制)问:要根据用户现在怎么写

    from rest_framework.throttling import SimpleRateThrottle
    class VisitThrottle(SimpleRateThrottle):
        scope = 'luffy'
        def get_cache_key(self, request, view):
            return self.get_ident(request)
    

    在setting里配置:(一分钟访问三次)

    REST_FRAMEWORK = {
        'DEFAULT_THROTTLE_RATES':{
            'luffy':'3/m'
        }
    }
    

    在视图类里使用

    throttle_classes = [MyThrottles,]
    

    错误信息的中文提示:

    class Course(APIView):
        authentication_classes = [TokenAuth, ]
        permission_classes = [UserPermission, ]
        throttle_classes = [MyThrottles,]
    
        def get(self, request):
            return HttpResponse('get')
    
        def post(self, request):
            return HttpResponse('post')
    
        def throttled(self, request, wait):
            from rest_framework.exceptions import Throttled
            class MyThrottled(Throttled):
                default_detail = '傻逼啊'
                extra_detail_singular = '还有 {wait} second.'
                extra_detail_plural = '出了 {wait} seconds.'
            raise MyThrottled(wait)
    

    内置频率限制类:

    undefined

    BaseThrottle是所有类的基类

    方法:def get_ident(self, request)获取标识,其实就是获取ip,自定义的需要继承它

    1.AnonRateThrottle:未登录用户ip限制,需要配合auth模块用

    2.SimpleRateThrottle:重写此方法,可以实现频率现在,不需要咱们手写上面自定义的逻辑

    3.UserRateThrottle:登录用户频率限制,这个得配合auth模块来用

    4.ScopedRateThrottle:应用在局部视图上的(忽略)

    四 内置频率类及全局使用

    REST_FRAMEWORK = {
        'DEFAULT_THROTTLE_CLASSES':['app01.utils.VisitThrottle',],
        'DEFAULT_THROTTLE_RATES':{
            'luffy':'3/m'
        }
    }
    

    五 源码分析

    def check_throttles(self, request):
    	for throttle in self.get_throttles():
    		if not throttle.allow_request(request, self):
    			self.throttled(request, throttle.wait())
    
    def throttled(self, request, wait):
    	#抛异常,可以自定义异常,实现错误信息的中文显示
    	raise exceptions.Throttled(wait)
    
    class SimpleRateThrottle(BaseThrottle):
        # 咱自己写的放在了全局变量,他的在django的缓存中
        cache = default_cache
        # 获取当前时间,跟咱写的一样
        timer = time.time
        # 做了一个字符串格式化,
        cache_format = 'throttle_%(scope)s_%(ident)s'
        scope = None
        # 从配置文件中取DEFAULT_THROTTLE_RATES,所以咱配置文件中应该配置,否则报错
        THROTTLE_RATES = api_settings.DEFAULT_THROTTLE_RATES
    
        def __init__(self):
            if not getattr(self, 'rate', None):
                # 从配置文件中找出scope配置的名字对应的值,比如咱写的‘3/m’,他取出来
                self.rate = self.get_rate()
            # 解析'3/m',解析成 3      m
            self.num_requests, self.duration = self.parse_rate(self.rate)
        # 这个方法需要重写
        def get_cache_key(self, request, view):
            """
            Should return a unique cache-key which can be used for throttling.
            Must be overridden.
    
            May return `None` if the request should not be throttled.
            """
            raise NotImplementedError('.get_cache_key() must be overridden')
    
        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:
                # 获取在setting里配置的字典中的之,self.scope是 咱写的luffy
                return self.THROTTLE_RATES[self.scope]
            except KeyError:
                msg = "No default throttle rate set for '%s' scope" % self.scope
                raise ImproperlyConfigured(msg)
        # 解析 3/m这种传参
        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)
            # 只取了第一位,也就是 3/mimmmmmmm也是代表一分钟
            duration = {'s': 1, 'm': 60, 'h': 3600, 'd': 86400}[period[0]]
            return (num_requests, duration)
        # 逻辑跟咱自定义的相同
        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()
        # 成功返回true,并且插入到缓存中
        def throttle_success(self):
            """
            Inserts the current request's timestamp along with the key
            into the cache.
            """
            self.history.insert(0, self.now)
            self.cache.set(self.key, self.history, self.duration)
            return True
        # 失败返回false
        def throttle_failure(self):
            """
            Called when a request to the API has failed due to throttling.
            """
            return False
    
        def wait(self):
            """
            Returns the recommended next request time in seconds.
            """
            if self.history:
                remaining_duration = self.duration - (self.now - self.history[-1])
            else:
                remaining_duration = self.duration
    
            available_requests = self.num_requests - len(self.history) + 1
            if available_requests <= 0:
                return None
    
            return remaining_duration / float(available_requests)
    
  • 相关阅读:
    消息中间件(一)MQ详解及四大MQ比较
    SIP协议
    PAT (Basic Level) Practice 1008 数组元素循环右移问题
    LeetCode-Algorithms 1. 两数之和
    PAT (Basic Level) Practice 1040 有几个PAT
    PAT (Basic Level) Practice 1023 组个最小数
    PAT (Basic Level) Practice 1021 个位数统计
    PAT (Basic Level) Practice 1007 素数对猜想
    PAT (Basic Level) Practice 1006 换个格式输出整数
    PAT (Basic Level) Practice 1004 成绩排名
  • 原文地址:https://www.cnblogs.com/Dr-wei/p/11842597.html
Copyright © 2011-2022 走看看