# -*- coding: utf-8 -*- """ Analytics Middleware =================== Middleware components for the Adtlas Analytics module. Provides request/response processing, analytics tracking, performance monitoring, and data collection capabilities. Middleware Components: - AnalyticsTrackingMiddleware: Core analytics tracking - PerformanceMonitoringMiddleware: Performance metrics collection - RequestLoggingMiddleware: Request/response logging - CacheMiddleware: Analytics-specific caching - SecurityMiddleware: Security monitoring and protection - RateLimitingMiddleware: API rate limiting - ErrorTrackingMiddleware: Error monitoring and reporting - DataCollectionMiddleware: Automated data collection Key Features: - Real-time analytics tracking - Performance metrics collection - Request/response logging - Cache management - Security monitoring - Rate limiting - Error tracking - Data validation - User behavior tracking - API usage analytics - Geographic tracking - Device detection - Session management Configuration: - Middleware ordering - Tracking settings - Performance thresholds - Cache configuration - Security rules - Rate limits - Logging levels Usage Examples: - Add to MIDDLEWARE setting in Django settings - Configure tracking parameters - Set performance thresholds - Define security rules - Configure rate limits Author: Adtlas Development Team Version: 1.0.0 Last Updated: 2024 """ import time import json import logging import hashlib from typing import Dict, Any, Optional, List, Callable, Union from datetime import datetime, timedelta from urllib.parse import urlparse from ipaddress import ip_address, ip_network from django.conf import settings from django.core.cache import cache from django.http import HttpRequest, HttpResponse, JsonResponse from django.utils import timezone from django.utils.deprecation import MiddlewareMixin from django.contrib.auth.models import AnonymousUser from django.contrib.gis.geoip2 import GeoIP2 from django.core.exceptions import ImproperlyConfigured from .models import ( Impression, PerformanceMetric, SfrAnalytics, BouyguesAnalytics ) from .constants import ( ANALYTICS_PROVIDERS, DEVICE_TYPES, CACHE_KEYS, CACHE_TIMEOUTS, PERFORMANCE_THRESHOLDS, RATE_LIMITS, HTTP_STATUS_CODES ) from .exceptions import ( AnalyticsException, RateLimitExceededException, ProcessingTimeoutException, SecurityException ) from .utils import ( get_client_ip, get_user_agent_info, validate_request_data, calculate_performance_score, generate_session_id ) # Configure logging logger = logging.getLogger(__name__) # ============================================================================= # BASE MIDDLEWARE # ============================================================================= class BaseAnalyticsMiddleware(MiddlewareMixin): """ Base middleware class for analytics components. Provides common functionality for all analytics middleware: - Configuration management - Logging integration - Error handling - Performance tracking - Context management Attributes: enabled: Whether middleware is enabled debug: Debug mode flag config: Middleware configuration start_time: Request start time context: Request context data """ def __init__(self, get_response: Callable): """ Initialize base analytics middleware. Args: get_response: Django get_response callable """ super().__init__(get_response) self.get_response = get_response # Load configuration self.enabled = getattr(settings, 'ANALYTICS_MIDDLEWARE_ENABLED', True) self.debug = getattr(settings, 'ANALYTICS_DEBUG', False) self.config = self._load_config() # Initialize components self._initialize_components() logger.info(f"{self.__class__.__name__} initialized") def _load_config(self) -> Dict[str, Any]: """ Load middleware configuration. Returns: Configuration dictionary """ default_config = { 'enabled': True, 'debug': False, 'log_level': 'INFO', 'cache_timeout': 300, 'max_retries': 3, 'timeout': 30 } # Get middleware-specific config config_key = f'{self.__class__.__name__.upper()}_CONFIG' middleware_config = getattr(settings, config_key, {}) # Merge with defaults config = {**default_config, **middleware_config} return config def _initialize_components(self) -> None: """ Initialize middleware components. Override in subclasses for specific initialization. """ pass def _should_process_request(self, request: HttpRequest) -> bool: """ Determine if request should be processed. Args: request: HTTP request object Returns: True if request should be processed """ if not self.enabled: return False # Skip for certain paths skip_paths = self.config.get('skip_paths', [ '/admin/', '/static/', '/media/', '/favicon.ico' ]) for path in skip_paths: if request.path.startswith(path): return False # Skip for certain methods skip_methods = self.config.get('skip_methods', ['OPTIONS']) if request.method in skip_methods: return False return True def _get_request_context(self, request: HttpRequest) -> Dict[str, Any]: """ Extract context information from request. Args: request: HTTP request object Returns: Request context dictionary """ context = { 'method': request.method, 'path': request.path, 'query_string': request.META.get('QUERY_STRING', ''), 'user_agent': request.META.get('HTTP_USER_AGENT', ''), 'referer': request.META.get('HTTP_REFERER', ''), 'client_ip': get_client_ip(request), 'timestamp': timezone.now().isoformat(), 'session_key': getattr(request.session, 'session_key', None), 'user_id': request.user.id if hasattr(request, 'user') and request.user.is_authenticated else None, 'is_authenticated': hasattr(request, 'user') and request.user.is_authenticated, 'content_type': request.content_type, 'content_length': request.META.get('CONTENT_LENGTH', 0) } # Add geographic information try: geo_info = self._get_geographic_info(context['client_ip']) context.update(geo_info) except Exception as e: logger.warning(f"Failed to get geographic info: {e}") # Add device information try: device_info = get_user_agent_info(context['user_agent']) context.update(device_info) except Exception as e: logger.warning(f"Failed to get device info: {e}") return context def _get_geographic_info(self, ip_address: str) -> Dict[str, Any]: """ Get geographic information from IP address. Args: ip_address: Client IP address Returns: Geographic information dictionary """ try: if not ip_address or ip_address in ['127.0.0.1', 'localhost']: return {} geoip = GeoIP2() city_data = geoip.city(ip_address) country_data = geoip.country(ip_address) return { 'country_code': country_data.get('country_code', ''), 'country_name': country_data.get('country_name', ''), 'city': city_data.get('city', ''), 'region': city_data.get('region', ''), 'latitude': city_data.get('latitude'), 'longitude': city_data.get('longitude'), 'timezone': city_data.get('time_zone', '') } except Exception as e: logger.debug(f"GeoIP lookup failed: {e}") return {} def _log_middleware_action( self, action: str, request: HttpRequest, context: Optional[Dict[str, Any]] = None, level: str = 'INFO' ) -> None: """ Log middleware action. Args: action: Action description request: HTTP request object context: Additional context level: Log level """ try: log_data = { 'middleware': self.__class__.__name__, 'action': action, 'path': request.path, 'method': request.method, 'user_id': getattr(request.user, 'id', None) if hasattr(request, 'user') else None } if context: log_data.update(context) getattr(logger, level.lower())(f"Middleware {action}", extra=log_data) except Exception as e: logger.error(f"Middleware logging failed: {e}") # ============================================================================= # ANALYTICS TRACKING MIDDLEWARE # ============================================================================= class AnalyticsTrackingMiddleware(BaseAnalyticsMiddleware): """ Core analytics tracking middleware. Tracks user interactions, page views, API calls, and other analytics events automatically. Features: - Automatic impression tracking - User behavior analytics - Session management - Event correlation - Real-time data collection """ def _initialize_components(self) -> None: """ Initialize tracking components. """ self.track_impressions = self.config.get('track_impressions', True) self.track_api_calls = self.config.get('track_api_calls', True) self.track_user_behavior = self.config.get('track_user_behavior', True) self.batch_size = self.config.get('batch_size', 100) self.flush_interval = self.config.get('flush_interval', 60) # Initialize batch storage self.impression_batch = [] self.last_flush = time.time() def process_request(self, request: HttpRequest) -> Optional[HttpResponse]: """ Process incoming request for analytics tracking. Args: request: HTTP request object Returns: None to continue processing """ if not self._should_process_request(request): return None try: # Set request start time request._analytics_start_time = time.time() # Generate or retrieve session ID session_id = self._get_or_create_session_id(request) request._analytics_session_id = session_id # Extract request context request._analytics_context = self._get_request_context(request) # Track impression if applicable if self.track_impressions and self._is_impression_request(request): self._track_impression(request) self._log_middleware_action('request_processed', request) except Exception as e: logger.error(f"Analytics tracking request processing failed: {e}") return None def process_response( self, request: HttpRequest, response: HttpResponse ) -> HttpResponse: """ Process response for analytics tracking. Args: request: HTTP request object response: HTTP response object Returns: Modified response object """ if not self._should_process_request(request): return response try: # Calculate response time if hasattr(request, '_analytics_start_time'): response_time = time.time() - request._analytics_start_time # Track API call if applicable if self.track_api_calls and self._is_api_request(request): self._track_api_call(request, response, response_time) # Track user behavior if self.track_user_behavior: self._track_user_behavior(request, response, response_time) # Flush batch if needed self._flush_batch_if_needed() self._log_middleware_action('response_processed', request, { 'status_code': response.status_code, 'content_length': len(response.content) if hasattr(response, 'content') else 0 }) except Exception as e: logger.error(f"Analytics tracking response processing failed: {e}") return response def _get_or_create_session_id(self, request: HttpRequest) -> str: """ Get or create analytics session ID. Args: request: HTTP request object Returns: Session ID string """ # Try to get from session if hasattr(request, 'session'): session_id = request.session.get('analytics_session_id') if session_id: return session_id # Generate new session ID session_id = generate_session_id(request) # Store in session if available if hasattr(request, 'session'): request.session['analytics_session_id'] = session_id return session_id def _is_impression_request(self, request: HttpRequest) -> bool: """ Determine if request should be tracked as impression. Args: request: HTTP request object Returns: True if request is an impression """ # Track page views and API endpoints impression_patterns = self.config.get('impression_patterns', [ '/api/campaigns/', '/api/channels/', '/api/creatives/' ]) return any(pattern in request.path for pattern in impression_patterns) def _is_api_request(self, request: HttpRequest) -> bool: """ Determine if request is an API call. Args: request: HTTP request object Returns: True if request is an API call """ return request.path.startswith('/api/') def _track_impression(self, request: HttpRequest) -> None: """ Track impression event. Args: request: HTTP request object """ try: impression_data = { 'session_id': getattr(request, '_analytics_session_id', ''), 'user_id': request.user.id if hasattr(request, 'user') and request.user.is_authenticated else None, 'path': request.path, 'method': request.method, 'client_ip': get_client_ip(request), 'user_agent': request.META.get('HTTP_USER_AGENT', ''), 'referer': request.META.get('HTTP_REFERER', ''), 'timestamp': timezone.now(), 'context': getattr(request, '_analytics_context', {}) } # Add to batch self.impression_batch.append(impression_data) logger.debug(f"Impression tracked: {request.path}") except Exception as e: logger.error(f"Impression tracking failed: {e}") def _track_api_call(self, request: HttpRequest, response: HttpResponse, response_time: float) -> None: """ Track API call analytics. Args: request: HTTP request object response: HTTP response object response_time: Response time in seconds """ try: # Determine provider based on path provider = None if '/sfr/' in request.path: provider = ANALYTICS_PROVIDERS.SFR elif '/bouygues/' in request.path: provider = ANALYTICS_PROVIDERS.BOUYGUES if provider: analytics_data = { 'provider': provider, 'endpoint': request.path, 'method': request.method, 'status_code': response.status_code, 'response_time': response_time, 'user_id': request.user.id if hasattr(request, 'user') and request.user.is_authenticated else None, 'session_id': getattr(request, '_analytics_session_id', ''), 'client_ip': get_client_ip(request), 'timestamp': timezone.now(), 'context': getattr(request, '_analytics_context', {}) } # Store analytics data based on provider if provider == ANALYTICS_PROVIDERS.SFR: SfrAnalytics.objects.create(**analytics_data) elif provider == ANALYTICS_PROVIDERS.BOUYGUES: BouyguesAnalytics.objects.create(**analytics_data) logger.debug(f"API call tracked: {request.path} ({response_time:.3f}s)") except Exception as e: logger.error(f"API call tracking failed: {e}") def _track_user_behavior(self, request: HttpRequest, response: HttpResponse, response_time: float) -> None: """ Track user behavior analytics. Args: request: HTTP request object response: HTTP response object response_time: Response time in seconds """ try: behavior_data = { 'session_id': getattr(request, '_analytics_session_id', ''), 'user_id': request.user.id if hasattr(request, 'user') and request.user.is_authenticated else None, 'action': self._determine_user_action(request), 'path': request.path, 'method': request.method, 'status_code': response.status_code, 'response_time': response_time, 'timestamp': timezone.now(), 'context': getattr(request, '_analytics_context', {}) } # Store in cache for batch processing cache_key = f"user_behavior_{request.user.id if hasattr(request, 'user') and request.user.is_authenticated else 'anonymous'}_{int(time.time())}" cache.set(cache_key, behavior_data, timeout=CACHE_TIMEOUTS.SHORT) logger.debug(f"User behavior tracked: {behavior_data['action']}") except Exception as e: logger.error(f"User behavior tracking failed: {e}") def _determine_user_action(self, request: HttpRequest) -> str: """ Determine user action from request. Args: request: HTTP request object Returns: User action string """ method_actions = { 'GET': 'view', 'POST': 'create', 'PUT': 'update', 'PATCH': 'update', 'DELETE': 'delete' } base_action = method_actions.get(request.method, 'unknown') # Add context based on path if '/campaigns/' in request.path: return f"{base_action}_campaign" elif '/channels/' in request.path: return f"{base_action}_channel" elif '/creatives/' in request.path: return f"{base_action}_creative" elif '/analytics/' in request.path: return f"{base_action}_analytics" return base_action def _flush_batch_if_needed(self) -> None: """ Flush impression batch if conditions are met. """ current_time = time.time() should_flush = ( len(self.impression_batch) >= self.batch_size or (current_time - self.last_flush) >= self.flush_interval ) if should_flush and self.impression_batch: self._flush_impression_batch() def _flush_impression_batch(self) -> None: """ Flush impression batch to database. """ try: # Create impression objects impressions = [ Impression(**data) for data in self.impression_batch ] # Bulk create Impression.objects.bulk_create(impressions, ignore_conflicts=True) logger.info(f"Flushed {len(self.impression_batch)} impressions") # Clear batch self.impression_batch.clear() self.last_flush = time.time() except Exception as e: logger.error(f"Impression batch flush failed: {e}") # Clear batch to prevent memory issues self.impression_batch.clear() # ============================================================================= # PERFORMANCE MONITORING MIDDLEWARE # ============================================================================= class PerformanceMonitoringMiddleware(BaseAnalyticsMiddleware): """ Performance monitoring middleware. Tracks response times, resource usage, and performance metrics for analytics optimization. Features: - Response time tracking - Resource usage monitoring - Performance threshold alerts - Slow query detection - Memory usage tracking """ def _initialize_components(self) -> None: """ Initialize performance monitoring components. """ self.track_response_times = self.config.get('track_response_times', True) self.track_database_queries = self.config.get('track_database_queries', True) self.track_cache_usage = self.config.get('track_cache_usage', True) self.slow_threshold = self.config.get('slow_threshold', 1.0) # seconds self.alert_threshold = self.config.get('alert_threshold', 5.0) # seconds def process_request(self, request: HttpRequest) -> Optional[HttpResponse]: """ Process request for performance monitoring. Args: request: HTTP request object Returns: None to continue processing """ if not self._should_process_request(request): return None try: # Set performance tracking attributes request._perf_start_time = time.time() request._perf_start_queries = self._get_query_count() request._perf_start_cache_hits = self._get_cache_hits() self._log_middleware_action('performance_tracking_started', request) except Exception as e: logger.error(f"Performance monitoring request processing failed: {e}") return None def process_response( self, request: HttpRequest, response: HttpResponse ) -> HttpResponse: """ Process response for performance monitoring. Args: request: HTTP request object response: HTTP response object Returns: Modified response object """ if not self._should_process_request(request): return response try: # Calculate performance metrics if hasattr(request, '_perf_start_time'): metrics = self._calculate_performance_metrics(request, response) # Store performance metrics if self.track_response_times: self._store_performance_metrics(request, response, metrics) # Check for performance issues self._check_performance_thresholds(request, metrics) # Add performance headers if debug mode if self.debug: self._add_performance_headers(response, metrics) self._log_middleware_action('performance_tracking_completed', request) except Exception as e: logger.error(f"Performance monitoring response processing failed: {e}") return response def _calculate_performance_metrics(self, request: HttpRequest, response: HttpResponse) -> Dict[str, Any]: """ Calculate performance metrics for request/response. Args: request: HTTP request object response: HTTP response object Returns: Performance metrics dictionary """ end_time = time.time() start_time = getattr(request, '_perf_start_time', end_time) metrics = { 'response_time': end_time - start_time, 'status_code': response.status_code, 'content_length': len(response.content) if hasattr(response, 'content') else 0, 'timestamp': timezone.now() } # Database query metrics if self.track_database_queries: start_queries = getattr(request, '_perf_start_queries', 0) end_queries = self._get_query_count() metrics['query_count'] = end_queries - start_queries # Cache usage metrics if self.track_cache_usage: start_cache_hits = getattr(request, '_perf_start_cache_hits', 0) end_cache_hits = self._get_cache_hits() metrics['cache_hits'] = end_cache_hits - start_cache_hits # Calculate performance score metrics['performance_score'] = calculate_performance_score(metrics) return metrics def _get_query_count(self) -> int: """ Get current database query count. Returns: Number of database queries """ try: from django.db import connection return len(connection.queries) except Exception: return 0 def _get_cache_hits(self) -> int: """ Get current cache hit count. Returns: Number of cache hits """ try: # This would need to be implemented based on cache backend # For now, return 0 return 0 except Exception: return 0 def _store_performance_metrics(self, request: HttpRequest, response: HttpResponse, metrics: Dict[str, Any]) -> None: """ Store performance metrics in database. Args: request: HTTP request object response: HTTP response object metrics: Performance metrics dictionary """ try: PerformanceMetric.objects.create( endpoint=request.path, method=request.method, response_time=metrics['response_time'], status_code=metrics['status_code'], query_count=metrics.get('query_count', 0), cache_hits=metrics.get('cache_hits', 0), content_length=metrics['content_length'], performance_score=metrics['performance_score'], user_id=request.user.id if hasattr(request, 'user') and request.user.is_authenticated else None, session_id=getattr(request, '_analytics_session_id', ''), timestamp=metrics['timestamp'] ) logger.debug(f"Performance metrics stored for {request.path}") except Exception as e: logger.error(f"Performance metrics storage failed: {e}") def _check_performance_thresholds(self, request: HttpRequest, metrics: Dict[str, Any]) -> None: """ Check performance metrics against thresholds. Args: request: HTTP request object metrics: Performance metrics dictionary """ response_time = metrics['response_time'] # Log slow requests if response_time > self.slow_threshold: logger.warning( f"Slow request detected: {request.path} ({response_time:.3f}s)", extra={ 'path': request.path, 'method': request.method, 'response_time': response_time, 'metrics': metrics } ) # Alert for very slow requests if response_time > self.alert_threshold: logger.error( f"Very slow request detected: {request.path} ({response_time:.3f}s)", extra={ 'path': request.path, 'method': request.method, 'response_time': response_time, 'metrics': metrics, 'alert': True } ) def _add_performance_headers(self, response: HttpResponse, metrics: Dict[str, Any]) -> None: """ Add performance headers to response. Args: response: HTTP response object metrics: Performance metrics dictionary """ try: response['X-Response-Time'] = f"{metrics['response_time']:.3f}s" response['X-Query-Count'] = str(metrics.get('query_count', 0)) response['X-Cache-Hits'] = str(metrics.get('cache_hits', 0)) response['X-Performance-Score'] = f"{metrics['performance_score']:.2f}" except Exception as e: logger.error(f"Adding performance headers failed: {e}") # ============================================================================= # RATE LIMITING MIDDLEWARE # ============================================================================= class RateLimitingMiddleware(BaseAnalyticsMiddleware): """ Rate limiting middleware for analytics APIs. Implements rate limiting to prevent abuse and ensure fair usage of analytics resources. Features: - Per-user rate limiting - Per-IP rate limiting - Endpoint-specific limits - Sliding window algorithm - Rate limit headers """ def _initialize_components(self) -> None: """ Initialize rate limiting components. """ self.default_limit = self.config.get('default_limit', RATE_LIMITS.DEFAULT) self.window_size = self.config.get('window_size', 3600) # 1 hour self.per_user_limits = self.config.get('per_user_limits', {}) self.per_endpoint_limits = self.config.get('per_endpoint_limits', {}) self.whitelist_ips = self.config.get('whitelist_ips', []) def process_request(self, request: HttpRequest) -> Optional[HttpResponse]: """ Process request for rate limiting. Args: request: HTTP request object Returns: Rate limit response or None to continue """ if not self._should_process_request(request): return None try: # Check if IP is whitelisted client_ip = get_client_ip(request) if client_ip in self.whitelist_ips: return None # Check rate limits rate_limit_result = self._check_rate_limits(request) if rate_limit_result['exceeded']: return self._create_rate_limit_response(rate_limit_result) # Update rate limit counters self._update_rate_limit_counters(request) self._log_middleware_action('rate_limit_checked', request, { 'remaining': rate_limit_result['remaining'], 'limit': rate_limit_result['limit'] }) except Exception as e: logger.error(f"Rate limiting request processing failed: {e}") return None def process_response( self, request: HttpRequest, response: HttpResponse ) -> HttpResponse: """ Process response for rate limiting headers. Args: request: HTTP request object response: HTTP response object Returns: Response with rate limit headers """ if not self._should_process_request(request): return response try: # Add rate limit headers rate_limit_info = self._get_rate_limit_info(request) self._add_rate_limit_headers(response, rate_limit_info) except Exception as e: logger.error(f"Rate limiting response processing failed: {e}") return response def _check_rate_limits(self, request: HttpRequest) -> Dict[str, Any]: """ Check if request exceeds rate limits. Args: request: HTTP request object Returns: Rate limit check result """ # Get rate limit key and limit rate_key = self._get_rate_limit_key(request) limit = self._get_rate_limit(request) # Get current usage current_usage = self._get_current_usage(rate_key) # Check if limit exceeded exceeded = current_usage >= limit remaining = max(0, limit - current_usage) # Calculate reset time reset_time = self._get_reset_time(rate_key) return { 'exceeded': exceeded, 'limit': limit, 'remaining': remaining, 'reset_time': reset_time, 'current_usage': current_usage } def _get_rate_limit_key(self, request: HttpRequest) -> str: """ Generate rate limit key for request. Args: request: HTTP request object Returns: Rate limit key string """ # Use user ID if authenticated, otherwise IP if hasattr(request, 'user') and request.user.is_authenticated: identifier = f"user_{request.user.id}" else: identifier = f"ip_{get_client_ip(request)}" # Include endpoint in key for endpoint-specific limits endpoint = self._normalize_endpoint(request.path) return f"rate_limit_{identifier}_{endpoint}" def _get_rate_limit(self, request: HttpRequest) -> int: """ Get rate limit for request. Args: request: HTTP request object Returns: Rate limit value """ # Check endpoint-specific limits endpoint = self._normalize_endpoint(request.path) if endpoint in self.per_endpoint_limits: return self.per_endpoint_limits[endpoint] # Check user-specific limits if hasattr(request, 'user') and request.user.is_authenticated: user_id = str(request.user.id) if user_id in self.per_user_limits: return self.per_user_limits[user_id] return self.default_limit def _normalize_endpoint(self, path: str) -> str: """ Normalize endpoint path for rate limiting. Args: path: Request path Returns: Normalized endpoint string """ # Remove IDs and normalize path import re normalized = re.sub(r'/\d+/', '/{id}/', path) normalized = re.sub(r'/[a-f0-9-]{36}/', '/{uuid}/', normalized) return normalized def _get_current_usage(self, rate_key: str) -> int: """ Get current usage for rate limit key. Args: rate_key: Rate limit key Returns: Current usage count """ try: return cache.get(rate_key, 0) except Exception: return 0 def _get_reset_time(self, rate_key: str) -> datetime: """ Get reset time for rate limit key. Args: rate_key: Rate limit key Returns: Reset time """ # Calculate next window start now = timezone.now() window_start = now.replace(minute=0, second=0, microsecond=0) next_window = window_start + timedelta(hours=1) return next_window def _update_rate_limit_counters(self, request: HttpRequest) -> None: """ Update rate limit counters for request. Args: request: HTTP request object """ try: rate_key = self._get_rate_limit_key(request) # Increment counter try: cache.add(rate_key, 0, timeout=self.window_size) cache.incr(rate_key) except ValueError: # Key doesn't exist, set to 1 cache.set(rate_key, 1, timeout=self.window_size) except Exception as e: logger.error(f"Rate limit counter update failed: {e}") def _create_rate_limit_response(self, rate_limit_result: Dict[str, Any]) -> JsonResponse: """ Create rate limit exceeded response. Args: rate_limit_result: Rate limit check result Returns: Rate limit response """ response_data = { 'error': True, 'error_code': 'RATE_LIMIT_EXCEEDED', 'message': 'Rate limit exceeded', 'details': { 'limit': rate_limit_result['limit'], 'remaining': rate_limit_result['remaining'], 'reset_time': rate_limit_result['reset_time'].isoformat() } } response = JsonResponse(response_data, status=429) self._add_rate_limit_headers(response, rate_limit_result) return response def _get_rate_limit_info(self, request: HttpRequest) -> Dict[str, Any]: """ Get rate limit information for request. Args: request: HTTP request object Returns: Rate limit information """ rate_key = self._get_rate_limit_key(request) limit = self._get_rate_limit(request) current_usage = self._get_current_usage(rate_key) remaining = max(0, limit - current_usage) reset_time = self._get_reset_time(rate_key) return { 'limit': limit, 'remaining': remaining, 'reset_time': reset_time, 'current_usage': current_usage } def _add_rate_limit_headers(self, response: HttpResponse, rate_limit_info: Dict[str, Any]) -> None: """ Add rate limit headers to response. Args: response: HTTP response object rate_limit_info: Rate limit information """ try: response['X-RateLimit-Limit'] = str(rate_limit_info['limit']) response['X-RateLimit-Remaining'] = str(rate_limit_info['remaining']) response['X-RateLimit-Reset'] = str(int(rate_limit_info['reset_time'].timestamp())) except Exception as e: logger.error(f"Adding rate limit headers failed: {e}") # ============================================================================= # EXPORTS # ============================================================================= # Export all middleware classes __all__ = [ 'BaseAnalyticsMiddleware', 'AnalyticsTrackingMiddleware', 'PerformanceMonitoringMiddleware', 'RateLimitingMiddleware', ]