DRF频率组件的使用方式

发表于 更新于 分类于 ， ， Changyan： 本文字数： 941 阅读时长 ≈ 3 分钟

DRF频率组件与上面两个组件也是几乎一样，只是用来做判断的逻辑不一样而已，下面使用自定义方式实现对ip地址进行访问频率控制

局部频率组件

自定义频率组件类

组件类位置：app.untils.app_throttle.VisitThottle

组件类代码：

1
2
3
4
5
6
7
8
9
# 引入Throttle类
from rest_framework.throttling import SimpleRateThrottle

class VisitThottle(SimpleRateThrottle):
    # 设定访问速度每分钟5次
    rate = '5/m'

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

rate代表访问评率，上面表示每分钟五次，get_cache_key是必须存在的，它的返回值告诉当前频率控制组件要使用什么方式区分访问者（比如ip地址）。

引入自定义频率组件类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 引入自定义访问速率类
from app.untils.app_throttle import VisitThottle

# 书籍列表
class BooksView(ModelViewSet):
    # 设置queryset和自定义序列化类
    queryset = BookModel.objects.all()
    serializer_class = BookSerialize
    # 需要认证的类
    authentication_classes = [UserAuth]
    # 需要判断权限的类测试
    permission_classes = [UserPermission]
    # 需要限制频率的组件类
    throttle_classes = [VisitThottle]

使用postman接口测试

1
2
3
4
# 如果一分钟访问超过5次，响应内容如下：
{
    "detail": "Request was throttled. Expected available in 59 seconds."
}

view.py所有代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
# DRF组件
from django.http import JsonResponse

from rest_framework.views import APIView
from rest_framework.viewsets import ModelViewSet
# from rest_framework.exceptions import APIException

# 引入用户模型
from app.models import User as UserModel
from app.models import UserToken as UserTokenModel
from app.models import Book as BookModel
from app.untils import get_token
# 引入自定义序列化类
from app.api_serialize import BookSerialize
# 引入自定义认证类
from app.untils.authentication import UserAuth
# 引入自定义权限类
from app.untils.permission import UserPermission
# 引入自定义访问速率类
from app.untils.app_throttle import VisitThottle

# 用户登录
class UserView(APIView):
    def post(self, request):
        # 定义一个返回字典
        response = dict()
        # 定义一个接收数据字典
        recev_data = {'username', 'password'}
        # 判断用户发送的post数据是否是要接受数据的子集，即用户发送的字段是否符合包含要接受的字段
        if recev_data.issubset(set(request.data)):
            # 接收用户发送符合要求的数据
            userInfo = dict()
            for item in recev_data:
                userInfo[item] = request.data.get(item)
            # 查询是否有该用户
            user_obj = UserModel.objects.filter(**userInfo).first()
            # 判断用户名密码是否正确
            if user_obj:
                # 生成access_token
                access_token = get_token.generater_token()
                # 将生成的access_token写入数据表
                UserTokenModel.objects.update_or_create(user=user_obj, defaults={
                    'token':access_token
                })
                response['status_code'] = 200
                response['status_message'] = '恭喜你，登录成功'
                response['access_token'] = access_token
                response['user_role'] = user_obj.get_level_display()
            else:
                response['status_code'] = 201
                response['status_message'] = '用户名或密码错误'
        return JsonResponse(response)

# 书籍列表
class BooksView(ModelViewSet):
    # 设置queryset和自定义序列化类
    queryset = BookModel.objects.all()
    serializer_class = BookSerialize
    # 需要认证的类
    authentication_classes = [UserAuth]
    # 需要判断权限的类测试
    permission_classes = [UserPermission]
    # 需要限制频率的组件类
    throttle_classes = [VisitThottle]

全局组件

创建自定义频率类

1
2
3
4
5
6
7
8
9
10
11
# 引入Throttle类
from rest_framework.throttling import SimpleRateThrottle

class VisitThottle(SimpleRateThrottle):
    # 设定访问速度每分钟5次
    # rate = '5/m'
    # 访问速率名称，setting中定义
    scope = 'visit_rate'

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

setting配置全局限制频率

1
2
3
4
5
6
7
8
9
10
11
# DRF全局认证
REST_FRAMEWORK = {
    # 全局认证类
    # 'DEFAULT_AUTHENTICATION_CLASSES' : ['app.untils.authentication.UserAuth'],
    # 全局访问频率限制类
    'DEFAULT_THROTTLE_CLASSES' : ['app.untils.app_throttle.VisitThottle'],
    # 频率限制速率，每分钟3次
    'DEFAULT_THROTTLE_RATES' : {
        'visit_rate' : '3/m'
    }
}

使用postman接口测试

1
2
3
4
# 如果一分钟访问超过5次，响应内容如下：
{
    "detail": "Request was throttled. Expected available in 59 seconds."
}

超过速率修改响应消息：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
import time
import math

from rest_framework import exceptions


class MyException(exceptions.Throttled):
    default_detail = '连接次数过多'
    extra_detail_plural = extra_detail_singular = '请在{wait}秒内访问'

    def __init__(self, wait=None, detail=None, code=None):
        super().__init__(wait=wait, detail=detail, code=code)


class VisitThrottle():
    user_visit_information = dict()
    visited_times = 1
    period = 60
    allow_times_per_minute = 5
    first_time_visit = True

    def allow_request(self, request, view):
        self.request_host = request_host = request.META.get("REMOTE_ADDR")
        current_user_info = self.user_visit_information.get(request_host, None)

        if not self.__class__.first_time_visit:
            self.user_visit_information[request_host][0] += 1
            current_visit_times = self.user_visit_information[request_host][0]

            if current_visit_times > self.allow_times_per_minute:
                if self._current_time - current_user_info[1] <= self.period:
                    if len(current_user_info) > 2:
                        current_user_info[2] = self._time_left
                    else:
                        current_user_info.append(self._time_left)

                    view.throttled = self.throttled
                    return None
                else:
                    self.__class__.first_time_visit = True

        if self.first_time_visit:
            self.__class__.first_time_visit = False
            self._initial_infomation()

        return True

    def wait(self):
        return self.period - self.user_visit_information[self.request_host][2]

    def throttled(self, request, wait):
        raise MyException(wait=wait)

    @property
    def _current_time(self):
        return time.time()

    @property
    def _time_left(self):
        return math.floor(self._current_time - self.user_visit_information.get(self.request_host)[1])

    def _initial_infomation(self):
        self.user_visit_information[self.request_host] = [self.visited_times, self._current_time]