中文字幕最新精品,玖玖国产精品视频,激情久久综合

主頁 > 知識庫 > Django REST framework 限流功能的使用

Django REST framework 限流功能的使用

正文開始

先說一個限流這個概念，最早接觸這個概念是在前端。真實的業務場景是在搜索框中輸入文字進行搜索時，并不希望每輸一個字符都去調用后端接口，而是有停頓后才真正的調用接口。這個功能很有必要，一方面減少前端請求與渲染的壓力，同時減輕后端接口訪問的壓力。類似前端的功能的代碼如下：

// 前端函數限流示例
function throttle(fn, delay) {
    var timer;
    return function () {
        var _this = this;
        var args = arguments;
        if (timer) {
            return;
        }
        timer = setTimeout(function () {
            fn.apply(_this, args);
            timer = null;
        }, delay)
    }
}

但是后端的限流從目的上來說與前端類似，但是實現上會有所不同，讓我們看看 DRF 的限流。

1. DRF 中的限流

項目配置

# demo/settings.py

REST_FRAMEWORK = {
    # ...
    'DEFAULT_THROTTLE_CLASSES': (
        'rest_framework.throttling.AnonRateThrottle',
        'rest_framework.throttling.UserRateThrottle',
         'rest_framework.throttling.ScopedRateThrottle',
    ),
    'DEFAULT_THROTTLE_RATES': {
        'anon': '10/day',
        'user': '2/day'
    },
}

# article/views.py

# 基于ViewSet的限流
class ArticleViewSet(viewsets.ModelViewSet, ExceptionMixin):
    """
    允許用戶查看或編輯的API路徑。
    """
    queryset = Article.objects.all()
    # 使用默認的用戶限流
    throttle_classes = (UserRateThrottle,)
    serializer_class = ArticleSerializer

# 基于view的限流
@throttle_classes([UserRateThrottle])

因為我配置的用戶每天只能請求兩次，所以在請求第三次之后就會給出 429 Too Many Requests的異常，具體的異常信息為下一次可用時間為 86398 秒后。

2. 限流進階配置

上述演示的限流配置適用于對用戶的限流，比如我換個用戶繼續訪問，依然是有兩次的機會。

$ curl -H 'Accept: application/json; indent=4' -u root:root   http://127.0.0.1:8000/api/article/1/ 
{
    "id": 1,
    "creator": "admin",
    "tag": "現代詩",
    "title": "如果",
    "content": "今生今世 永不再將你想起\n除了\n除了在有些個\n因落淚而濕潤的夜里 如果\n如果你愿意"
}

分別介紹一下三種限流類

AnonRateThrottle 適用于任何用戶對接口訪問的限制
UserRateThrottle 適用于請求認證結束后對接口訪問的限制
ScopedRateThrottle 適用于對多個接口訪問的限制

所以三種不同的類適用于不同的業務場景，具體使用根據不同的業務場景選擇，通過配置相對應 scope 的頻率的配置就可以達到預期的效果。

3. 限流思路分析

試想一下如果是你編碼實現這個需求應該怎么實現？

其實這個功能不難，核心的參數就是時間、次數、使用范圍，下面演示對函數調用次數的限制。

from functools import wraps

TOTAL_RATE = 2

FUNC_SCOPE = ['test', 'test1']


def rate_count(func):
    func_num = {
        # 需要注意函數名不能重復
        func.__name__: 0
    }

    @wraps(func)
    def wrapper():
        if func.__name__ in FUNC_SCOPE:
            if func_num[func.__name__] >= TOTAL_RATE:
                raise Exception(f"{func.__name__}函數調用超過設定次數")
            result = func()
            func_num[func.__name__] += 1
            print(f" 函數 {func.__name__} 調用次數為： {func_num[func.__name__]}")
            return result
        else:
            # 不在計數限制的函數不受限制
            return func()

    return wrapper


@rate_count
def test1():
    pass


@rate_count
def test2():
    print("test2")
    pass


if __name__ == "__main__":
    try:
        test2()
        test2()
        test1()
        test1()
        test1()
    except Exception as e:
        print(e)
    test2()
    test2()
    
"""
test2
test2
 函數 test1 調用次數為： 1
 函數 test1 調用次數為： 2
test1函數調用超過設定次數
test2
test2
"""

這里實現了對函數調用次數的監控同時設置了能夠使用該功能的函數。當函數調用次數超過設定閥值久拋出異常。只是這里沒有對時間做限制。

4. 源碼分析

剛才分析了如何實現對函數調用次數的限制，對于一個請求來說可能會復雜一點，下面就看看 DRF 如何實現的：

class SimpleRateThrottle(BaseThrottle):
   
    # ......
    
    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()

        # 根據設置時間的限制改變請求次數的緩存
        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()
    
    # ......
    
class UserRateThrottle(SimpleRateThrottle):
    """
    Limits the rate of API calls that may be made by a given user.

    The user id will be used as a unique cache key if the user is
    authenticated.  For anonymous requests, the IP address of the request will
    be used.
    """
    scope = 'user'

    def get_cache_key(self, request, view):
        if request.user.is_authenticated:
            ident = request.user.pk
        else:
            # 考慮到用戶沒有認證的情況 與 AnonRateThrottle 中 key 一致
            ident = self.get_ident(request)
        # 根據設置的范圍構建緩存的 key
        return self.cache_format % {
            'scope': self.scope,
            'ident': ident
        }

綜上所述：