Django REST framework使用及源码分析之节流

节流，访问次数控制。

views.py

class OrderView(APIView):
    '''
    订单相关业务，返回订单字典。
    添加验证功能，验证是否已登录。
    '''
    # authentication_classes = [Authtication,]
    # permission_classes = [MyPermission1, ]
    throttle_classes=[VisitThrottle,]
    '''该刚就是添加Authtication认证，可以添加多个，如果是一个空列表，则不认证，也就是可以跳过认证'''
    def get(self,request,*args,**kwargs):
        ret = {'code':1000,'msg':None}
        try:
            ret['data'] = ORDER_DICT
        except Exception as e:
            pass
        return  JsonResponse(ret)

增加一个局部设置，throttle_classes

 

访问控制类：

import time
VISIT_RECORD={}

class VisitThrottle(object):
    '''
    节流控制，返回True为通过访问，False为限制访问。
    '''
    def allow_request(self,request,view):
        remote_addr = request.META.get('REMOTE_ADDR') #获取访问的IP
        print(remote_addr)
        ctime = time.time()
        if remote_addr not in VISIT_RECORD:  #判断ip是存在，不存在则把ip和时间记录，然后通过访问。
            VISIT_RECORD[remote_addr] = [ctime,]

            return True
        history = VISIT_RECORD.get(remote_addr)  #如果ip存在，则找到这个对应的时间
        print(VISIT_RECORD)
        self.history = history
        while history and history[-1] < ctime - 10:
            #循环，找到这个ip对应的时间列表，如果最早的历史时间小于当前时间减去10秒，那么将其删除
            history.pop()
        if len(history) <3:
            '''如果列表长度小于3，则将当前时间添加到列表索引0的位置，也就是最前面'''
            history.insert(0,ctime)
            return True
    def wait(self):
        '''还有多久可以访问
        通过用限制时间减去（当前时间减去历史最早时间），将时间返回，系统将自动发给前端。'''
        ctime = time.time()
        return 10 - (ctime - self.history[-1])

全局设置：

REST_FRAMEWORK = {
    # 全局使用的认证类
    "DEFAULT_AUTHENTICATION_CLASSES":['app.utils.auth.Authtication', ],
    "UNAUTHENTICATED_USER":None, # 匿名，request.user = None
    "UNAUTHENTICATED_TOKEN":None,# 匿名，request.auth = None
    "DEFAULT_PERMISSION_CLASSES":['app.utils.permission.MyPermission1'],
    "DEFAULT_THROTTLE_CLASSES":["app.utils.throttle.VisitThrottle"],
}

源码流程梳理：

 

基本和验证、权限差不多。

 

除了上面这种自定义的节流控制，REST framework有个自带的节流控制，比较方便。

节流类：

from rest_framework.throttling import BaseThrottle,SimpleRateThrottle
class VisitThrottle(SimpleRateThrottle):
    scope = "Luffy"

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


class UserThrottle(SimpleRateThrottle):
    scope = "LuffyUser"

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

只需实现get_cache_key，配置文件添加配置即可。

 

scope只是一个字符串，当做字典的key

get_ident 是针对匿名用户的，根据Ip控制。

request.user.username 通过用户名控制。

 

 

 

配置：

REST_FRAMEWORK = {
    # 全局使用的认证类
    "DEFAULT_AUTHENTICATION_CLASSES":['api.utils.auth.FirstAuthtication','api.utils.auth.Authtication', ],
    # "DEFAULT_AUTHENTICATION_CLASSES":['api.utils.auth.FirstAuthtication', ],
    # "UNAUTHENTICATED_USER":lambda :"匿名用户"
    "UNAUTHENTICATED_USER":None, # 匿名，request.user = None
    "UNAUTHENTICATED_TOKEN":None,# 匿名，request.auth = None
    "DEFAULT_PERMISSION_CLASSES":['api.utils.permission.SVIPPermission'],
    "DEFAULT_THROTTLE_CLASSES":["api.utils.throttle.UserThrottle"],
    "DEFAULT_THROTTLE_RATES":{
        "Luffy":'3/m',
        "LuffyUser":'10/m',
    }
}

上面配置，需要些字典的key:时间/单位。