Троттлинг — Django REST Framework

Троттлинг в Django REST Framework

Троттлинг (throttling) ограничивает частоту запросов от клиентов. Это защищает API от злоупотреблений, DDoS-атак и перегрузки сервера.

#Что такое троттлинг

Throttling — механизм ограничения количества запросов от клиента за определённый промежуток времени. В отличие от прав доступа, троттлинг применяется ко всем пользователям, включая авторизованных.

#Настройка троттлинга

Глобальная настройка в settings.py:

REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_CLASSES': [
        'rest_framework.throttling.AnonRateThrottle',
        'rest_framework.throttling.UserRateThrottle',
    ],
    'DEFAULT_THROTTLE_RATES': {
        'anon': '100/day',      # Анонимные пользователи
        'user': '1000/day',     # Авторизованные пользователи
        'auth': '50/hour',      # Для endpoints аутентификации
        'burst': '10/min',      # Краткосрочный лимит
        'sustained': '1000/day', # Долгосрочный лимит
    }
}

Формат rates:

'100/day' — 100 запросов в день
'10/min' — 10 запросов в минуту
'1000/hour' — 1000 запросов в час
'5/second' — 5 запросов в секунду

#Встроенные классы троттлинга

#AnonRateThrottle

Ограничивает запросы от анонимных пользователей:

from rest_framework.throttling import AnonRateThrottle

class MyView(APIView):
    throttle_classes = [AnonRateThrottle]
    
    def get(self, request):
        return Response({'message': 'Hello'})

Ключ: IP-адрес клиента
Rate: 'anon' из DEFAULT_THROTTLE_RATES

#UserRateThrottle

Ограничивает запросы от авторизованных пользователей:

from rest_framework.throttling import UserRateThrottle

class MyView(APIView):
    throttle_classes = [UserRateThrottle]
    
    def get(self, request):
        return Response({'message': 'Hello'})

Ключ: ID пользователя (или IP для анонимных)
Rate: 'user' из DEFAULT_THROTTLE_RATES

#ScopedRateThrottle

Ограничивает запросы по scope (области применения):

# settings.py
REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_CLASSES': [
        'rest_framework.throttling.ScopedRateThrottle',
    ],
    'DEFAULT_THROTTLE_RATES': {
        'articles': '100/hour',
        'comments': '10/minute',
    }
}

# views.py
class ArticleViewSet(viewsets.ModelViewSet):
    queryset = Article.objects.all()
    serializer_class = ArticleSerializer
    throttle_scope = 'articles'  # Применяет лимит 'articles'

Ключ: Комбинация scope + пользователь/IP
Rate: Значение из DEFAULT_THROTTLE_RATES по ключу throttle_scope

#Кастомный класс троттлинга

Создание своего класса:

from rest_framework.throttling import SimpleRateThrottle

class BurstRateThrottle(SimpleRateThrottle):
    """Краткосрочный троттлинг — 10 запросов в минуту"""
    scope = 'burst'
    
    def get_cache_key(self, request, view):
        if request.user and request.user.is_authenticated:
            ident = request.user.pk
        else:
            ident = self.get_ident(request)  # IP-адрес
        
        return self.cache_format % {
            'scope': self.scope,
            'ident': ident
        }

# settings.py
REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_CLASSES': [
        'myapp.throttling.BurstRateThrottle',
    ],
    'DEFAULT_THROTTLE_RATES': {
        'burst': '10/min',
    }
}

#Троттлинг для конкретных view

#Отключение троттлинга

from rest_framework.throttling import ScopedRateThrottle

class HighPriorityView(APIView):
    throttle_classes = []  # Отключить троттлинг
    
    def get(self, request):
        return Response({'message': 'No throttling'})

#Разные лимиты для разных view

# views.py
from rest_framework.throttling import UserRateThrottle

class ArticleList(APIView):
    throttle_classes = [UserRateThrottle]
    # Использует 'user' rate из settings
    
    def get(self, request):
        return Response(Article.objects.all())

class CommentCreate(APIView):
    throttle_classes = [UserRateThrottle]
    
    def get_throttles(self):
        # Переопределить rate для этого view
        from rest_framework.throttling import ScopedRateThrottle
        return [ScopedRateThrottle()]
    
    throttle_scope = 'comments'
    
    def post(self, request):
        return Response({'status': 'created'})

#Кастомизация ключа троттлинга

Можно изменить способ идентификации клиента:

from rest_framework.throttling import UserRateThrottle

class XForwardedForUserRateThrottle(UserRateThrottle):
    """Использовать X-Forwarded-For IP вместо remote_addr"""
    
    def get_cache_key(self, request, view):
        if request.user and request.user.is_authenticated:
            ident = request.user.pk
        else:
            # Получить реальный IP из заголовка
            ident = request.META.get('HTTP_X_FORWARDED_FOR', '')
            if not ident:
                ident = self.get_ident(request)
        
        return self.cache_format % {
            'scope': self.scope,
            'ident': ident
        }

#Ответ при превышении лимита

При превышении лимита DRF возвращает:

HTTP/1.1 429 Too Many Requests
Content-Type: application/json
X-RateLimit-Limit: 100
X-RateLimit-Remaining: 0
X-RateLimit-Reset: 1677721600

{
    "detail": "Request was throttled. Expected available in 3600 seconds."
}

Заголовки:

X-RateLimit-Limit — максимальное количество запросов
X-RateLimit-Remaining — оставшееся количество запросов
X-RateLimit-Reset — время сброса лимита (Unix timestamp)

#Хранение счётчиков

По умолчанию DRF использует кэш Django для хранения счётчиков:

# settings.py
CACHES = {
    'default': {
        'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
        # Для production используйте Redis или Memcached:
        # 'BACKEND': 'django.core.cache.backends.redis.RedisCache',
        # 'LOCATION': 'redis://127.0.0.1:6379/1',
    }
}

Важно: Для production используйте Redis или Memcached, иначе лимиты не будут работать корректно при нескольких серверах.

#Троттлинг для аутентификации

Защита endpoints аутентификации от брутфорса:

# settings.py
REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_RATES': {
        'auth': '5/hour',  # Только 5 попыток входа в час
    }
}

# views.py
from rest_framework.throttling import ScopedRateThrottle

class LoginView(APIView):
    throttle_classes = [ScopedRateThrottle]
    throttle_scope = 'auth'
    
    def post(self, request):
        # Логика аутентификации
        pass

#Комбинация троттлов

Можно использовать несколько классов троттлинга одновременно:

class ArticleViewSet(viewsets.ModelViewSet):
    queryset = Article.objects.all()
    serializer_class = ArticleSerializer
    throttle_classes = [
        BurstRateThrottle,      # 10/min
        SustainedRateThrottle,  # 1000/day
    ]

Логика: Запрос блокируется, если превышен любой из лимитов.

#Best Practices

Всегда включайте троттлинг для публичных API
Разные лимиты для анонимов и пользователей — анонимам меньше
Защищайте endpoints аутентификации — 5-10 попыток в час
Используйте Redis/Memcached для кэша в production
Мониторьте превышения — логируйте 429 ответы
Документируйте лимиты — укажите в документации API
Consider exponential backoff — увеличивайте блокировку при повторных нарушениях