四、REST framework的三大組件之一：限流

1、介紹

1.1、開發過程中，某個接口不想用戶訪問過于頻繁，使用限流機制進行控制，比如：現在1分鐘發5次等。

1.2、限流時，需要基于某個條件生成唯一標識進行限制，比如獲取訪問對象的IP、用戶信息的主鍵、ID、用戶名等等進行限制，

2、限制示例，1分鐘訪問5次"

步驟一：唯一標識"： [12:00:30，12:01:05，12:00:50，12:00:20，12:00:05]，列表為訪問記錄

步驟二：獲取當前時間 12:00:15

步驟三：當前時間減去1分鐘=計數時間, 12:01:15 - 60s<1分鐘> = 12:00:45

步驟四：通過計數時間在唯一標識的列表中，找到少于 12:00:45的時間，并進行刪除，得到列表 [12:00:30，12:01:05，12:00:50]

步驟五：計數等到列表的長度，如果列表長度 > 5<次>，錯誤限制訪問，反之，正常訪問。

3、在ext中新建myThrottle.py，并添加限流類

# 導入rest_framework的限流類
from rest_framework.throttling import BaseThrottle


class MyThrottle(BaseThrottle):

    # 繼承BaseThrottle，需要重寫allow_request方法
    def allow_request(self, request, view):
        """
        Return `True` if the request should be allowed, `False` otherwise.
        """
        # 返回True 表示允許訪問
        return True

4、限流類BaseThrottle的源碼解釋

class BaseThrottle:
    """
    Rate throttling of requests.
    """
　　# 限流入口方法
    def allow_request(self, request, view):
        """
        Return `True` if the request should be allowed, `False` otherwise.
        """
        raise NotImplementedError('.allow_request() must be overridden')

    # 獲取IP地址，用于后續構建唯一標識
    def get_ident(self, request):
        """
        Identify the machine making the request by parsing HTTP_X_FORWARDED_FOR
        if present and number of proxies is > 0. If not use all of
        HTTP_X_FORWARDED_FOR if it is available, if not use REMOTE_ADDR.
        """
        xff = request.META.get('HTTP_X_FORWARDED_FOR')
        remote_addr = request.META.get('REMOTE_ADDR')
        num_proxies = api_settings.NUM_PROXIES

        if num_proxies is not None:
            if num_proxies == 0 or xff is None:
                return remote_addr
            addrs = xff.split(',')
            client_addr = addrs[-min(num_proxies, len(addrs))]
            return client_addr.strip()

        return ''.join(xff.split()) if xff else remote_addr

    # 等待時間
    def wait(self):
        """
        Optionally, return a recommended number of seconds to wait before
        the next request.
        """
        return None

5、通過限流類BaseThrottle，可以看出我們要使用基類的話必須重寫它的三個方法，但是DRF為了避免我們進行復雜操作，提供了SimpleRateThrottle類，修改MyThrottle類

# 導入rest_framework的限流類
from rest_framework.throttling import SimpleRateThrottle


class MyThrottle(SimpleRateThrottle):

    # 使用SimpleRateThrottle需要重寫get_cache_key方法，get_cache_key生成唯一標識
    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')

6、在認證時，我們對request.user進行了賦值，及表示數據庫表的實例，因此，這里我們可以獲取用戶的唯一ID，修改方法get_cache_key

# 導入rest_framework的限流類
from rest_framework.throttling import SimpleRateThrottle


class MyThrottle(SimpleRateThrottle):

    # 定義scope
    scope = "xxx"

    # 使用SimpleRateThrottle需要重寫get_cache_key方法，get_cache_key生成唯一標識
    # 目的：防止唯一標識重復
    def get_cache_key(self, request, view):
        # 判斷是否有request.user
        if request.user:
            # 如果有request.user，獲取唯一標識，及主鍵
            ident = request.user.pk
        else:
            # 如果沒有，調用父類的get_ident，獲取請求用戶的IP<在request的請求頭中查找>
            ident = self.get_ident(request)
        # 對self.cache_format進行，字符串格式化 cache_format = 'throttle_%(scope)s_%(ident)s'
        # 調用父類的self.scope，父類中的scope = None，因此可以自己進行定義
        return self.cache_format % {
            'scope': self.scope,
            'ident': ident
        }

7、在 SimpleRateThrottle的__init__方法，通過初始化獲取了 self.rate = self.get_rate()，get_rate()源碼介紹

    def get_rate(self):
        """
        Determine the string representation of the allowed request rate.
        """
        # 反射判斷是否有scope，沒有拋出異常
        if not getattr(self, 'scope', None):
            msg = ("You must set either `.scope` or `.rate` for '%s' throttle" %
                   self.__class__.__name__)
            raise ImproperlyConfigured(msg)

        # 驗證是否有THROTTLE_RATES屬性，沒有拋出異常
        try:
            # 沒有定義THROTTLE_RATES時，通過配置，獲取全局設置的THROTTLE_RATES，當前沒有進行全局設置，不手動設置會拋出異常
            # 獲取配置信息，如幾秒訪問幾次 {"xxx": "5/s"}
            # self.THROTTLE_RATES[self.scope]是通過self.scope為鍵，獲取值，因此配置時THROTTLE_RATES的鍵必須和scope一致
            return self.THROTTLE_RATES[self.scope]
        except KeyError:
            msg = "No default throttle rate set for '%s' scope" % self.scope
            raise ImproperlyConfigured(msg)

8、通過get_rate()源碼我們知道還需要定義 THROTTLE_RATES屬性，修改get_cache_key，添加THROTTLE_RATES配置

class MyThrottle(SimpleRateThrottle):

    # 定義scope
    scope = "xxx"
    # 配置THROTTLE_RATES，鍵與scope保持一致
    THROTTLE_RATES = {"xxx": "10/m"}

    # 使用SimpleRateThrottle需要重寫get_cache_key方法，get_cache_key生成唯一標識
    # 目的：防止唯一標識重復
    def get_cache_key(self, request, view):
        # 判斷是否有request.user
        if request.user:
            # 如果有request.user，獲取唯一標識，及主鍵
            ident = request.user.pk
        else:
            # 如果沒有，調用父類的get_ident，獲取請求用戶的IP<在request的請求頭中查找>
            ident = self.get_ident(request)
        # 對self.cache_format進行，字符串格式化 cache_format = 'throttle_%(scope)s_%(ident)s'
        # 調用父類的self.scope，父類中的scope = None，因此可以自己進行定義
        return self.cache_format % {
            'scope': self.scope,
            'ident': ident
        }

9、在SimpleRateThrottle源碼中，cache = default_cache為調用Redis，并把訪問記錄放在Redis中，因此需要設置Redis連接，配置setting

CACHES = {
    "default": {
        "BACKEND": "django_redis.cache.RedisCache",
        "LOCATION": "redis://IP:端口/數據庫編號",  # Redis服務器地址和端口，以及使用的數據庫編號（默認為0，這里改為1）
        "OPTIONS": {
            "CLIENT_CLASS": "django_redis.client.DefaultClient",
            "CONNECTION_POOL_KWARGS": {"max_connections": 100},
            "DECODE_RESPONSES": True,  # 自動將Redis存儲的字節串解碼為字符串
        }
    }
}

10，配置視圖函數，并使用postman進行接口訪問

class LoginView(MyAPIView):

    # authentication_classes 為認證類的默認配置參數，必須怎么寫，詳細過程可以查看源碼
    # 當不使用認證類時，列表為空
    # 請注意，當類中使用authentication_classes=[認證類1, 認證類2, ...]時，認證默認使用就近原則，使用類中的authentication_classes
    authentication_classes = [QueryParamsAuthentication]  # 通過url認證
    # 請注意，當類中使用permission_classes=[權限類1, 權限類2, ...]時，權限默認使用就近原則，使用類中的permission_classes
    permission_classes = [MyPermission1, MyPermission2]  # role=1和2的的權限
    # 應用限流類，throttle_classes限流類的默認配置參數，必須怎么寫，詳細過程可以查看源碼
    throttle_classes = [MyThrottle]

    def get(self, request, *args, **kwargs):
        return Response('GET')

# 接口返回： 前5次，正常返回
# 當訪問超過五次時，返回
# {
#     "detail": "請求超過了限速。 Expected available in 56 seconds."
# }

11、結合源碼可以看出，限流類也支持全局配置使用，使用方法請自行查看源碼，但是這種功能我們基本不在全局使用。

12、在第8個步驟中，如果不定義THROTTLE_RATES，它會從配置文件中查找，因此我們可以在全局配置THROTTLE_RATES屬性，配置完成后，刪除MyThrottle中的THROTTLE_RATES，接口訪問正常

REST_FRAMEWORK = {
    "UNAUTHENTICATED_USER": None,
    # 添加DEFAULT_PERMISSION_CLASSES使其支持全局認證
    "DEFAULT_AUTHENTICATION_CLASSES": ["auto_project.ext.myAuthentication.QueryParamsAuthentication",
                                       "auto_project.ext.myAuthentication.HeaderAuthentication",
                                       "auto_project.ext.myAuthentication.BodyAuthentication",
                                       "auto_project.ext.myAuthentication.NoAuthentication",
                                       ],
    # 添加DEFAULT_PERMISSION_CLASSES，所有視圖默認全部應用權限
    "DEFAULT_PERMISSION_CLASSES": ["auto_project.ext.MyPermission.MyPermission1"],
    # 添加限流配置
    "DEFAULT_THROTTLE_RATES": {"xxx": "5/m", "yyy": "10/m"},
    # 版本控制
    "VERSION_PARAM": "version",
    "DEFAULT_VERSION": "v1",
    "ALLOWED_VERSIONS": ["v1", "v2"],
    "NON_FIELD_ERRORS_KEY": "全局驗證"
}

13、源碼執行流程，請注意類對象實例化時，先觸發__init__方法

13.1、限流調用流程： as_view -> dispatch，在dispatch中通過 self.initial(request, *args, **kwargs)，調用限流方法check_throttles

    def check_throttles(self, request):
        """
        Check if request should be throttled.
        Raises an appropriate exception if the request is throttled.
        """
        # 創建空列表
        throttle_durations = []
        # 循環throttle_classes類對象
        for throttle in self.get_throttles():
            # 獲取對象后，調用allow_request，not True = False， 當為True時，才執行下面的流程
            # allow_request，函數入口方法
            if not throttle.allow_request(request, self):
                # 調用throttle.wait()，相當于還需要等待多長時間
                # wait執行流程：當前時間減去最早的訪問記錄，然后使用間隔時間減去得到的值，得到還需要等待多久
                throttle_durations.append(throttle.wait())

        # 如果超出限流現在，調用if流程
        if throttle_durations:
            # Filter out `None` values which may happen in case of config / rate
            # changes, see #1438
            # 刪除None
            durations = [
                duration for duration in throttle_durations
                if duration is not None
            ]
            # 獲取最大限流，如，使用了很多限流，取其中最大的限流
            duration = max(durations, default=None)
            # 找到視圖中的throttled方法，拋出異常
            self.throttled(request, duration)

14、定制返回code和描述信息

14.1、根據源碼到拋出異常時，調用的是self.throttled(request, duration)方法，該方法在APIview中定義

    def throttled(self, request, wait):
        """
        If request is throttled, determine what kind of exception to raise.
        """
        raise exceptions.Throttled(wait)

14.2、self.throttled(request, duration)方法調用exceptions.Throttled(wait)

class Throttled(APIException):
    status_code = status.HTTP_429_TOO_MANY_REQUESTS
    default_detail = _('Request was throttled.')
    extra_detail_singular = _('Expected available in {wait} second.')
    extra_detail_plural = _('Expected available in {wait} seconds.')
    default_code = 'throttled'

    def __init__(self, wait=None, detail=None, code=None):
        if detail is None:
            detail = force_str(self.default_detail)
        if wait is not None:
            wait = math.ceil(wait)
            detail = ' '.join((
                detail,
                force_str(ngettext(self.extra_detail_singular.format(wait=wait),
                                   self.extra_detail_plural.format(wait=wait),
                                   wait))))
        self.wait = wait
        super().__init__(detail, code)

14.3、exceptions.Throttled只是對detail, code進行了設置，那么我們可以重寫throttled方法，而后自定義我們自己的Throttled<在之前我們定義的MyAPIView中重寫>

    # 重寫 APIView的throttled方法
    def throttled(self, request, wait):
        # 返回自定義異常返回值
        raise Throttled(wait)


# 導入相PIException異常類和math標準數學庫math
from rest_framework.exceptions import APIException
import math


# 編寫自己的Throttled
class Throttled(APIException):

    # __init__方法
    def __init__(self, wait=None):
        if wait is not None:
            # wait不為空時，去整數
            wait = math.ceil(wait)
        # 賦值
        self.wait = wait
        # 調用父類的__init__方法，只對當前類做處理，后續邏輯還是通過APIException處理
        super().__init__({'code': 100005, 'msg': f'請求超過次數，請等待{wait}s后，進行訪問'})

14,4、再次訪問接口時，超出限制后，接口返回

總結：

1、限流不要在全局應用

2、重寫限流時的返回結果，有助于定制后續的狀態碼

3、限流可以結合認證進行使用

posted @ 2024-12-26 18:48 蝸牛·哥閱讀(55) 評論(0) 收藏舉報

蝸牛·哥

四、REST framework的三大組件之一：限流

公告