Django DRF限流组件

本文介绍: 在DRF中，限流发生在认证、权限之后，限流组件的使用步骤： 1、编写自定义限流类；2、在set t in g s.py中配置 redis；3、安装 django–redis;4、启动 redis 服务；5、局部应用，一般是在核心的视图中使用，不会全局使用。一、自定义限流类,t hrot tle.py，设计了 2个限流类，一个是针对匿名用户的限流，匿名用户的唯一标识选择IP地址；一个针对登录用户的限流，登录用户的唯一标识是用户名。二、全局配置，set t in g s.py。三、局部应用，view s.py。

在DRF中，限流发生在认证、权限之后，限流组件的使用步骤： 1、编写自定义限流类； 2、在set t in g s.py中配置 redis； 3、安装 django–redis; 4、启动 redis 服务； 5、局部应用，一般是在核心的视图中使用，不会全局使用。限流组件的应用案例如下：

一、自定义限流类,t hrottle.py，设计了 2个限流类，一个是针对匿名用户的限流，匿名用户的唯一标识选择IP地址；一个针对登录用户的限流，登录用户的唯一标识是用户名。

from rest_framework.throttling import SimpleRateThrottle
from django.core.cache import cache as default_cache

# 限流组件，匿名用户访问，没有登录的用户，肯定是没有user的，直接获取IP地址
class IpThrottle(SimpleRateThrottle):
    scope = "ip"
    # 局部配置，一分钟访问10次；也可以配置到全局；
    # THROTTLE_RATES = {"ip": "10/m"}
    cache = default_cache  # default_cache 会读取配置文件中redis缓存的配置

    def get_cache_key(self, request, view):
        # 获取请求用户的IP地址（去request中找请求头）
        ident = self.get_ident(request)
        return self.cache_format % {'scope': self.scope, 'ident': ident}


# 限流组件，用户限流类
class UserThrottle(SimpleRateThrottle):
    scope = "user"
    # 局部配置，一分钟访问5次；也可以配置到全局；
    # THROTTLE_RATES = {"user": "5/m"}
    cache = default_cache  # default_cache 会读取配置文件中redis缓存的配置

    def get_cache_key(self, request, view):
        ident = request.user.pk  #用户ID
        return self.cache_format % {'scope': self.scope, 'ident': ident}

二、全局配置，settin g s.py

REST_FRAMEWORK = {
    # 限流全局配置
    "DEFAULT_THROTTLE_RATES":{
        "ip":"10/m",
        "user":"5/m",
    }
}

三、局部应用，views.py

from ext.throttle import IpThrottle,UserThrottle



class LoginView(APIView):
    # login页面不需要认证就可以登录，所以单独设置为空；
    authentication_classes = []
    permission_classes = []
    # 应用限流组件,使用IP限流
    throttle_classes = [IpThrottle,]

    def post(self,request):
        # 1、接收用户提交的用户名和密码；
        user = request.data.get("username")
        pwd = request.data.get("password")
        # 2、数据库校验；
        user_object = models.UserInfo.objects.filter(username=user,password=pwd).first()
        if not user_object:
            return Response({"status":False,"msg":"用户名或者密码错误"})
        # 用户名密码正确为用户生产token
        token = str(uuid.uuid4())
        user_object.token = token
        user_object.save()
        return Response({"status":True,"msg":"登录成功！","token":token})


class AvatarView(NbApiView):
    # 老板或者员工可以访问
    permission_classes = [UserPermission,BossPermission]
    # 对登录用户使用登录用户限流
    throttle_classes = [UserThrottle,]

    def get(self,request):
        return Response({"status":True,"data":[11,22,33,44]})