Django REST Framework用户层级API限流异常及自定义响应需求

在Django REST Framework中基于用户层级（免费层、付费层、高级付费层）设置API调用限制时遇到两个问题：

1. 无论用户所属层级如何，均触发默认的3次/分钟限流规则
2. 希望为不同层级用户定制限流触发后的响应信息

提供的测试代码如下：

throttling.py #

from rest_framework.throttling import UserRateThrottle

class TierBasedRateThrottle(UserRateThrottle):
    def allow_request(self, request, view):
        user = request.user
        if user.is_authenticated:
            if user.tier_level == 'tier_1':
                self.rate = '4/minute'  # Rate limit for tier 1 users
            elif user.tier_level == 'tier_2':
                self.rate = '6/minute'  # Rate limit for tier 2 users
            elif user.tier_level == 'tier_3':
                self.rate = '10/minute'  # Rate limit for tier 3 users

        return super().allow_request(request, view)

settings.py #

REST_FRAMEWORK = {

'DEFAULT_FILTER_BACKENDS': ('django_filters.rest_framework.DjangoFilterBackend',),

# Use Django's standard `django.contrib.auth` permissions,
# or allow read-only access for unauthenticated users.
'DEFAULT_PERMISSION_CLASSES': [
    'rest_framework.permissions.DjangoModelPermissionsOrAnonReadOnly',
    'rest_framework.permissions.IsAuthenticated'
],
'DEFAULT_RENDERER_CLASSES': [
    'rest_framework.renderers.JSONRenderer',
    'rest_framework.renderers.BrowsableAPIRenderer',
],
'DEFAULT_PARSER_CLASSES': [
'rest_framework.parsers.JSONParser',
'rest_framework.parsers.FormParser',
'rest_framework.parsers.MultiPartParser'
],
'DEFAULT_AUTHENTICATION_CLASSES':[
'knox.auth.TokenAuthentication',
],
'DEFAULT_THROTTLE_CLASSES': [
    'accounts.throttling.TierBasedRateThrottle',
],
'DEFAULT_THROTTLE_RATES': {
    'user': '3/minute',
    'tier1': '5/minute',  # Default limit for tier1 users
    'tier2': '10/minute',  # Limit for tier2 users
    'tier3': '15/minute',  # Limit for tier3 users
},
}

views.py #

@api_view(['POST']) 
@authentication_classes([TokenAuthentication])
@permission_classes((IsAuthenticated, Tier2Permissions))
@throttle_classes([TierBasedRateThrottle])
def LiveSummary(request):
    queryset=liveSummary()
    return Response(queryset)

1. 修复层级限流不生效问题 #

问题根源在于UserRateThrottle的初始化逻辑：父类在构造时已经通过self.scope读取了settings中的默认限流速率（'user'对应的3/minute），并计算了num_requests和duration。直接在allow_request中修改self.rate不会自动更新这两个关键参数，导致限流依然使用默认值。

修改TierBasedRateThrottle类，在设置self.rate后手动更新num_requests和duration，同时从settings中读取对应层级的配置：

from rest_framework.throttling import UserRateThrottle
from rest_framework.settings import api_settings

class TierBasedRateThrottle(UserRateThrottle):
    def allow_request(self, request, view):
        user = request.user
        if user.is_authenticated:
            # 从settings的DEFAULT_THROTTLE_RATES中读取对应层级的速率
            tier_rate_map = {
                'tier_1': api_settings.DEFAULT_THROTTLE_RATES.get('tier1'),
                'tier_2': api_settings.DEFAULT_THROTTLE_RATES.get('tier2'),
                'tier_3': api_settings.DEFAULT_THROTTLE_RATES.get('tier3'),
            }
            self.rate = tier_rate_map.get(user.tier_level, api_settings.DEFAULT_THROTTLE_RATES['user'])
            # 重新计算num_requests和duration
            self.num_requests, self.duration = self.parse_rate(self.rate)
        
        return super().allow_request(request, view)

2. 定制限流触发后的响应信息 #

重写throttled方法，根据用户层级返回不同的响应内容：

from rest_framework.throttling import UserRateThrottle
from rest_framework.settings import api_settings
from rest_framework.exceptions import Throttled

class TierBasedRateThrottle(UserRateThrottle):
    def allow_request(self, request, view):
        user = request.user
        if user.is_authenticated:
            tier_rate_map = {
                'tier_1': api_settings.DEFAULT_THROTTLE_RATES.get('tier1'),
                'tier_2': api_settings.DEFAULT_THROTTLE_RATES.get('tier2'),
                'tier_3': api_settings.DEFAULT_THROTTLE_RATES.get('tier3'),
            }
            self.rate = tier_rate_map.get(user.tier_level, api_settings.DEFAULT_THROTTLE_RATES['user'])
            self.num_requests, self.duration = self.parse_rate(self.rate)
        
        return super().allow_request(request, view)
    
    def throttled(self, request, wait):
        user = request.user
        # 根据用户层级定制错误消息
        if user.tier_level == 'tier_1':
            detail = f"免费层用户请求过于频繁，请{int(wait)}秒后重试。如需更高限额，请升级至付费层。"
        elif user.tier_level == 'tier_2':
            detail = f"付费层用户请求过于频繁，请{int(wait)}秒后重试。如需更高限额，请升级至高级付费层。"
        elif user.tier_level == 'tier_3':
            detail = f"高级付费层用户请求过于频繁，请{int(wait)}秒后重试。"
        else:
            detail = super().throttled(request, wait).detail
        
        raise Throttled(detail=detail, wait=wait)

3. 额外注意事项 #

• 确保用户模型中tier_level字段的值与代码中的判断值（'tier_1'、'tier_2'、'tier_3'）完全匹配，避免大小写或拼写错误导致匹配失败。
• 如果不需要全局启用该限流类，可以移除settings.py中DEFAULT_THROTTLE_CLASSES的配置，仅在需要的视图上通过@throttle_classes装饰器指定，避免影响其他接口。

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