""" Scoped Rate Throttling Provides rate limiting with different scopes for different operation types """ from rest_framework.throttling import ScopedRateThrottle from django.conf import settings import logging logger = logging.getLogger(__name__) class DebugScopedRateThrottle(ScopedRateThrottle): """ Scoped rate throttle that can be bypassed in debug mode Usage: class MyViewSet(viewsets.ModelViewSet): throttle_scope = 'planner' throttle_classes = [DebugScopedRateThrottle] """ def allow_request(self, request, view): """ Check if request should be throttled Bypasses throttling if: - DEBUG mode is True - IGNY8_DEBUG_THROTTLE environment variable is True - User belongs to aws-admin or other system accounts - User is admin/developer role """ # Check if throttling should be bypassed debug_bypass = getattr(settings, 'DEBUG', False) env_bypass = getattr(settings, 'IGNY8_DEBUG_THROTTLE', False) # Bypass for system account users (aws-admin, default-account, etc.) system_account_bypass = False if hasattr(request, 'user') and request.user and hasattr(request.user, 'is_authenticated') and request.user.is_authenticated: try: # Check if user is in system account (aws-admin, default-account, default) if hasattr(request.user, 'is_system_account_user') and request.user.is_system_account_user(): system_account_bypass = True # Also bypass for admin/developer roles elif hasattr(request.user, 'is_admin_or_developer') and request.user.is_admin_or_developer(): system_account_bypass = True except (AttributeError, Exception): # If checking fails, continue with normal throttling pass if debug_bypass or env_bypass or system_account_bypass: # In debug mode or for system accounts, still set throttle headers but don't actually throttle # This allows testing throttle headers without blocking requests if hasattr(self, 'get_rate'): # Set headers for debugging self.scope = getattr(view, 'throttle_scope', None) if self.scope: # Get rate for this scope rate = self.get_rate() if rate: # Parse rate (e.g., "10/min") num_requests, duration = self.parse_rate(rate) # Set headers request._throttle_debug_info = { 'scope': self.scope, 'rate': rate, 'limit': num_requests, 'duration': duration } return True # Normal throttling behavior return super().allow_request(request, view) def get_rate(self): """ Get rate for the current scope """ if not self.scope: return None # Get throttle rates from settings throttle_rates = getattr(settings, 'REST_FRAMEWORK', {}).get('DEFAULT_THROTTLE_RATES', {}) # Get rate for this scope rate = throttle_rates.get(self.scope) # Fallback to default if scope not found if not rate: rate = throttle_rates.get('default', '100/min') return rate def parse_rate(self, rate): """ Parse rate string (e.g., "10/min") into (num_requests, duration) Returns: tuple: (num_requests, duration_in_seconds) """ if not rate: return None, None try: num, period = rate.split('/') num_requests = int(num) # Parse duration period = period.strip().lower() if period == 'sec' or period == 's': duration = 1 elif period == 'min' or period == 'm': duration = 60 elif period == 'hour' or period == 'h': duration = 3600 elif period == 'day' or period == 'd': duration = 86400 else: # Default to seconds duration = 1 return num_requests, duration except (ValueError, AttributeError): # Invalid rate format, default to 100/min logger.warning(f"Invalid rate format: {rate}, defaulting to 100/min") return 100, 60 def throttle_success(self): """ Called when request is allowed Sets throttle headers on response """ # This is called by DRF after allow_request returns True # Headers are set automatically by ScopedRateThrottle pass