能否统计Django REST Framework限流类接收的请求数量?
当然可以统计UserRateThrottle接收的请求数量!结合你使用的认证和权限配置,下面给你几个实用的方案:
方案1:自定义限流类,实时获取当前计数
默认的UserRateThrottle是通过Django缓存存储请求计数的,我们可以扩展它,在处理请求时把当前计数暴露出来,甚至记录到日志/数据库:
from rest_framework.throttling import UserRateThrottle from django.core.cache import cache class TrackedUserRateThrottle(UserRateThrottle): def allow_request(self, request, view): # 先调用父类逻辑判断是否允许请求 allowed = super().allow_request(request, view) # 构造当前用户的限流缓存key(和父类逻辑一致) user_id = request.user.pk cache_key = self.cache_format % { 'scope': self.scope, 'ident': user_id } # 从缓存中读取已使用的请求数 current_requests = cache.get(cache_key, 0) # 把计数挂载到request对象,方便后续视图使用 request.current_throttle_usage = current_requests return allowed
然后在你的视图中配置这个自定义限流类,并返回计数信息:
from rest_framework.views import APIView from rest_framework.response import Response class MyProtectedAPI(APIView): authentication_classes = [TokenAuthentication] permission_classes = [IsAuthenticated] throttle_classes = [TrackedUserRateThrottle] def get(self, request): # 获取已使用的请求数和限流上限 used = getattr(request, 'current_throttle_usage', 0) limit = int(self.throttle_classes[0].rate.split('/')[0]) return Response({ 'used_requests': used, 'daily_limit': limit, 'remaining_requests': limit - used })
方案2:直接读取缓存中的计数
如果你不想修改限流类,也可以直接根据UserRateThrottle的缓存规则,手动构造key读取计数:
from django.core.cache import cache from rest_framework.throttling import UserRateThrottle def get_user_throttle_count(user, scope="user"): """获取指定用户在当前周期内的限流请求数""" throttle = UserRateThrottle() throttle.scope = scope # 构造缓存key cache_key = throttle.cache_format % { 'scope': throttle.scope, 'ident': user.pk } return cache.get(cache_key, 0)
比如在视图中调用:
def get(self, request): used = get_user_throttle_count(request.user) return Response({"used_requests": used})
方案3:持久化统计(记录历史请求)
如果需要长期统计用户的请求趋势,可以在自定义限流类中添加数据库日志记录:
首先创建一个日志模型:
from django.db import models from django.contrib.auth import get_user_model User = get_user_model() class ThrottleRequestLog(models.Model): user = models.ForeignKey(User, on_delete=models.CASCADE) scope = models.CharField(max_length=50, default="user") request_time = models.DateTimeField(auto_now_add=True) class Meta: indexes = [ models.Index(fields=["user", "scope", "request_time"]), ]
然后修改自定义限流类:
class TrackedUserRateThrottle(UserRateThrottle): def allow_request(self, request, view): allowed = super().allow_request(request, view) # 记录每一次请求 ThrottleRequestLog.objects.create( user=request.user, scope=self.scope ) return allowed
之后就可以通过查询模型统计任意时间段的请求数:
from django.utils import timezone def get_user_daily_requests(user): today = timezone.now().date() return ThrottleRequestLog.objects.filter( user=user, request_time__date=today ).count()
注意事项
- 默认的
UserRateThrottle缓存会根据你设置的rate自动过期(比如1000/day会在每天零点重置计数) - 如果使用Redis等分布式缓存,上述方案依然有效,无需额外修改
内容的提问来源于stack exchange,提问作者user12742293
相关产品推荐
相关产品推荐

