How does Django REST framework implement built in access frequency control

  • 2021-07-26 08:16:41
  • OfStack

IP is used to control the access frequency for anonymous users, and user name is used to control the access frequency for logged-in users.


from rest_framework.throttling import SimpleRateThrottle

class VisitThrottle(SimpleRateThrottle):
  """ Anonymous user access frequency limit """
  scope = "AnonymousUser" #  Written casually, it can be used as key Save in cache 

  def get_cache_key(self, request, view):
    return self.get_ident(request)

class UserThrottle(SimpleRateThrottle):
  """ Access frequency limit of logged-in user """
  scope = "LoginUser"

  def get_cache_key(self, request, view):return request.user

redis can be configured


CACHES = {
  "default": {
    "BACKEND": "django_redis.cache.RedisCache",
    "LOCATION": "redis://127.0.0.1:6379",
    "OPTIONS": {
      "CLIENT_CLASS": "django_redis.client.DefaultClient",
      "CONNECTION_POOL_KWARGS": {"max_connections": 100}
      # "PASSWORD": " Password ",
    }
  }
}

The access frequency limit of anonymous users is set under the whole station here, as follows:


REST_FRAMEWORK = {
  "DEFAULT_THROTTLE_CLASSES": ["appxx.utils.VisitThrottle"],
  "DEFAULT_THROTTLE_RATES":{
      "AnonymousUser": "3/m", #  Anonymous user 1 Minutes can be accessed 3 Times, seconds (s) , points (m) , hour (h) , day (d)
      "LoginUser": "10/m", #  Login user 1 Minutes can be accessed 10 Times 
    }
}

The access frequency of the logged-in user is set in a separate view, and the view relies on authentication to tell whether the user is logged in, so the setting is as follows:


class BookViewSet(viewsets.ModelViewSet):
  authentication_classes = [TokenAuthentication]
  throttle_classes = [UserThrottle]
  queryset = models.Book.objects.all()
  serializer_class = serializers.BookSerializer

User authentication is as follows:


from rest_framework import authentication
from rest_framework import exceptionsfrom appxx import models

class TokenAuthentication(authentication.BaseAuthentication):
  """ Identity authentication """
  def authenticate(self, request):
    token = request.GET.get("token")
    obj = models.UserAuthToken.objects.filter(token=token).first()
    if not obj:
      raise exceptions.AuthenticationFailed(" Verification failed! ")
    else:
      return (obj.user.username, obj.token)

Related articles: