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)