首页
学习
活动
专区
圈层
工具
发布
社区首页 >问答首页 >Skip休息节流计数器

问Skip休息节流计数器
EN

Stack Overflow用户
提问于 2022-04-16 18:33:05
回答 1查看 122关注 0票数 1

当用户请求无效或服务器无法完成进程时,如何防止Django rest节流计数请求?

例如,我需要来自用户的params,但是当用户不给params时,Django rest节流仍然计算它。

当请求不成功时,是否有跳过节流计数器的解决方案?

示例

代码语言:javascript
复制
class OncePerHourAnonThrottle(AnonRateThrottle):
    rate = "1/hour"


class Autoliker(APIView):
    throttle_classes = [OncePerHourAnonThrottle]

    def get(self, request):
        content = {"status": "get"}
        return Response(content)

    def post(self, request):
        post_url = request.POST.get("url", None)
        print(post_url)

        content = {"status": "post"}
        return Response(content)

    def throttled(self, request, wait):
        raise Throttled(
            detail={
                "message": "request limit exceeded",
                "availableIn": f"{wait} seconds",
                "throttleType": "type",
            }
        )
EN

回答 1

Stack Overflow用户

发布于 2022-07-04 11:52:44

您可以为此创建一个装饰器。

代码语言:javascript
复制
class OncePerHourAnonThrottle(AnonRateThrottle):
    rate = "1/hour"

    def allow_request(self, request, view):
        """
        This function is copy of SimpleRateThrottle.allow_request
        The only difference is, instead of executing self.throttle_success
        it directly returns True and doesn't mark this request as success yet.
        """
        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()

        # Drop any requests from the history which have now passed the
        # throttle duration
        while self.history and self.history[-1] <= self.now - self.duration:
            self.history.pop()
        if len(self.history) >= self.num_requests:
            return False
        return True

def rate_limiter(view_function):
    @wraps(view_function)
    def inner(view_obj, request, *args, **kwargs):
        throttle = OncePerHourAnonThrottle()
        allowed = throttle.allow_request(request, None)
        if not allowed:
            raise exceptions.Throttled(throttle.wait())
        try:
            response = view_function(view_obj, request, *args, **kwargs)
        except Exception as exc:
            response = view_obj.handle_exception(exc)
        if response.status_code == 200:
            # now if everything goes OK, count this request as success
            throttle.throttle_success()
        return response

    return inner

class Autoliker(APIView):
   
    @rate_limiter
    def post(requests):
        # view logic
        pass

这是你怎么做的基本想法,现在你可以把它变成一个通用装饰器,甚至是基于类的装饰器。

票数 0
EN
页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接:

https://stackoverflow.com/questions/71896496

复制
相关文章

相似问题

领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档