DRF throttle (APIView)

Using Django Forms & APIs
1

hi
is there any way that I throttle a request base on specific status code?
as you see in code, i throttle all request no matter what is the status code.
but i want to the throttle incoming request if all the latest request ends with 404 status code not any other status code

class CheckUserPhone(APIView):
    throttle_scope = "check_phone"
    throttle_classes = (ScopedRateThrottle,)
    serializer_class = CheckUserPhoneSerializer

    def post(self, request):
        serializer = self.serializer_class(data=request.data)
        serializer.is_valid(raise_exception=True)
        phone = get_phone_from_serializer(serializer)

        user = User.objects.filter(phone=phone)
        if user.exist():
            return Response({"message": LoginSituation.LOGIN_REQUIRED}, status=status.HTTP_200_OK)
        sms_provider_result, code = generate_otp_and_send(phone)
        if sms_provider_result:
            save_otp_inside_cache(phone, code)
            return Response(
                {"message": LoginSituation.REGISTER_REQUIRED}, status=status.HTTP_404_NOT_FOUND
            )
        return Response({"message": FaultCode.SMS_PROVIDER_FAILURE}, status=status.HTTP_400_BAD_REQUEST)
2

You can create a custom throttle class. I adapted a sample code from my old projects to your code as follows. You can review it.

Custom Throttle Class

from rest_framework.throttling import BaseThrottle
from django.core.cache import cache
from django.utils import timezone
from datetime import timedelta

class StatusCodeThrottle(BaseThrottle):
    cache_key_prefix = 'status_code_throttle_'
    rate = 5  
    duration = 60  

    def get_cache_key(self, request):
        return f"{self.cache_key_prefix}{request.user.id}"

    def allow_request(self, request, view):
        cache_key = self.get_cache_key(request)
        request_data = cache.get(cache_key, [])
        now = timezone.now()
        
        request_data = [entry for entry in request_data if entry > now - timedelta(seconds=self.duration)]
        cache.set(cache_key, request_data, timeout=self.duration)
        
        if len(request_data) >= self.rate:
            return False
        
        return True

    def record_status_code(self, request, status_code):
        if status_code == 404:
            cache_key = self.get_cache_key(request)
            request_data = cache.get(cache_key, [])
            request_data.append(timezone.now())
            cache.set(cache_key, request_data, timeout=self.duration)

Using the Throttle in CheckUserPhoneView

from rest_framework.response import Response
from rest_framework import status
from rest_framework.views import APIView
from .throttles import StatusCodeThrottle

class CheckUserPhone(APIView):
    throttle_classes = (StatusCodeThrottle,)
    serializer_class = CheckUserPhoneSerializer

    def post(self, request):
        throttle = StatusCodeThrottle()
        if not throttle.allow_request(request, self):
            return Response({"detail": "Request limit reached"}, status=status.HTTP_429_TOO_MANY_REQUESTS)

        serializer = self.serializer_class(data=request.data)
        serializer.is_valid(raise_exception=True)
        phone = get_phone_from_serializer(serializer)
        
        user = User.objects.filter(phone=phone)
        if user.exists():
            response = Response({"message": LoginSituation.LOGIN_REQUIRED}, status=status.HTTP_200_OK)
        else:
            sms_provider_result, code = generate_otp_and_send(phone)
            if sms_provider_result:
                save_otp_inside_cache(phone, code)
                response = Response({"message": LoginSituation.REGISTER_REQUIRED}, status=status.HTTP_404_NOT_FOUND)
            else:
                response = Response({"message": FaultCode.SMS_PROVIDER_FAILURE}, status=status.HTTP_400_BAD_REQUEST)

        throttle.record_status_code(request, response.status_code)
        return response
1 Like
3

thank you, i wrote some code like that but not as good as your code

4

You are welcome, if you have a problem, feel free to tag me and I will be happy to help.

1 Like