Django DRF如何重写Custom throttles实现自定义节流规则

• 第一版继承BaseThrottle的实现没有任何限流逻辑，allow_request固定返回True等于完全放开所有请求，自然不会触发拦截。另外wait是节流类需要单独实现的实例方法，不能作为allow_request的入参定义。
• 第二版继承AnonRateThrottle重写parse_rate的实现有两个核心错误：一是parse_rate的入参是配置中传入的字符串格式限流规则（如10/s），直接传入元组(10,20)会导致后续调用split方法时抛出类型错误；二是原生AnonRateThrottle实现的是常规滑动窗口限流，和你需要的触发阈值后强制封禁固定时长的惩罚性规则逻辑完全不同，就算修正parse_rate写法也达不到预期效果。

你需要的是惩罚式节流规则：1秒内请求数超过10次时，直接拦截该用户后续所有请求20秒。直接继承BaseThrottle，借助Django缓存存储访问计数和封禁标记即可实现，完整代码如下：

import time
from django.core.cache import cache
from rest_framework.throttling import BaseThrottle


class CustomPunishThrottle(BaseThrottle):
    # 限流规则配置
    MAX_REQUESTS = 10       # 检查窗口内最大请求数
    CHECK_WINDOW = 1        # 频次检查窗口长度，单位：秒
    BAN_TIME = 20           # 触发阈值后的封禁时长，单位：秒

    def get_user_ident(self, request):
        """生成请求用户的唯一标识，登录用户用用户ID，匿名用户用客户端IP"""
        if request.user and request.user.is_authenticated:
            return f"throttle:user:{request.user.id}"
        # 获取客户端真实IP，兼容反向代理场景
        xff = request.META.get("HTTP_X_FORWARDED_FOR")
        if xff:
            client_ip = xff.split(",")[0].strip()
        else:
            client_ip = request.META.get("REMOTE_ADDR")
        return f"throttle:ip:{client_ip}"

    def allow_request(self, request, view):
        self.user_ident = self.get_user_ident(request)
        current_ts = time.time()

        # 第一步：检查当前用户是否处于封禁期
        ban_cache_key = f"{self.user_ident}:ban"
        ban_expire_ts = cache.get(ban_cache_key)
        if ban_expire_ts and ban_expire_ts > current_ts:
            self.retry_after = ban_expire_ts - current_ts
            return False
        # 封禁已过期则删除标记
        cache.delete(ban_cache_key)

        # 第二步：统计检查窗口内的有效请求数
        record_cache_key = f"{self.user_ident}:records"
        history_ts = cache.get(record_cache_key, [])
        # 过滤掉窗口外的过期请求记录
        valid_history = [ts for ts in history_ts if ts > current_ts - self.CHECK_WINDOW]

        # 第三步：判断是否触发限流阈值
        if len(valid_history) >= self.MAX_REQUESTS:
            # 触发阈值，设置封禁标记
            new_ban_expire = current_ts + self.BAN_TIME
            cache.set(ban_cache_key, new_ban_expire, self.BAN_TIME)
            # 清空历史请求记录，封禁结束后重新计数
            cache.delete(record_cache_key)
            self.retry_after = self.BAN_TIME
            return False

        # 未触发阈值，记录本次请求时间
        valid_history.append(current_ts)
        cache.set(record_cache_key, valid_history, self.CHECK_WINDOW + 1)
        return True

    def wait(self):
        """返回429响应需要等待的秒数，DRF会自动写入Retry-After响应头"""
        return self.retry_after

配置方式 #

• 全局生效：在项目DRF配置中添加该节流类

REST_FRAMEWORK = {
    "DEFAULT_THROTTLE_CLASSES": [
        "你的项目节流类存放路径.CustomPunishThrottle",
    ],
}

• 单视图生效：直接在视图类中指定throttle_classes属性即可

from rest_framework.views import APIView

class YourView(APIView):
    throttle_classes = [CustomPunishThrottle]
    # 其他视图逻辑

逻辑说明 #

• 所有请求优先校验封禁状态，封禁期内直接拦截，避免重复计数
• 仅对非封禁状态的请求做窗口内频次统计，达到阈值立即写入固定时长的封禁标记
• 缓存自动过期清理历史记录和封禁标记，不需要额外做垃圾数据清理
• 自动兼容登录用户和匿名用户的限流识别，反向代理部署场景也能正确获取客户端IP

内容的提问来源于stack exchange，提问作者boyenec