""" Error tracking utilities for Aniworld API. This module provides error tracking, logging, and reporting functionality for comprehensive error monitoring and debugging. """ import logging import uuid from datetime import datetime, timezone from typing import Any, Dict, Optional logger = logging.getLogger(__name__) class ErrorTracker: """ Centralized error tracking and management. Collects error metadata and provides insights into error patterns. """ def __init__(self): """Initialize error tracker.""" self.error_history: list[Dict[str, Any]] = [] self.max_history_size = 1000 def track_error( self, error_type: str, message: str, request_path: str, request_method: str, user_id: Optional[str] = None, status_code: int = 500, details: Optional[Dict[str, Any]] = None, request_id: Optional[str] = None, ) -> str: """ Track an error occurrence. Args: error_type: Type of error message: Error message request_path: Request path that caused error request_method: HTTP method user_id: User ID if available status_code: HTTP status code details: Additional error details request_id: Request ID for correlation Returns: Unique error tracking ID """ error_id = str(uuid.uuid4()) timestamp = datetime.now(timezone.utc).isoformat() error_entry = { "id": error_id, "timestamp": timestamp, "type": error_type, "message": message, "request_path": request_path, "request_method": request_method, "user_id": user_id, "status_code": status_code, "details": details or {}, "request_id": request_id, } self.error_history.append(error_entry) # Keep history size manageable if len(self.error_history) > self.max_history_size: self.error_history = self.error_history[-self.max_history_size:] logger.info( f"Error tracked: {error_id}", extra={ "error_id": error_id, "error_type": error_type, "status_code": status_code, "request_path": request_path, }, ) return error_id def get_error_stats(self) -> Dict[str, Any]: """ Get error statistics from history. Returns: Dictionary containing error statistics """ if not self.error_history: return {"total_errors": 0, "error_types": {}} error_types: Dict[str, int] = {} status_codes: Dict[int, int] = {} for error in self.error_history: error_type = error["type"] error_types[error_type] = error_types.get(error_type, 0) + 1 status_code = error["status_code"] status_codes[status_code] = status_codes.get(status_code, 0) + 1 return { "total_errors": len(self.error_history), "error_types": error_types, "status_codes": status_codes, "last_error": ( self.error_history[-1] if self.error_history else None ), } def get_recent_errors(self, limit: int = 10) -> list[Dict[str, Any]]: """ Get recent errors. Args: limit: Maximum number of errors to return Returns: List of recent error entries """ return self.error_history[-limit:] if self.error_history else [] def clear_history(self) -> None: """Clear error history.""" self.error_history.clear() logger.info("Error history cleared") # Global error tracker instance _error_tracker: Optional[ErrorTracker] = None def get_error_tracker() -> ErrorTracker: """ Get or create global error tracker instance. Returns: ErrorTracker instance """ global _error_tracker if _error_tracker is None: _error_tracker = ErrorTracker() return _error_tracker def reset_error_tracker() -> None: """Reset error tracker for testing.""" global _error_tracker _error_tracker = None class RequestContextManager: """ Manages request context for error tracking. Stores request metadata for error correlation. """ def __init__(self): """Initialize context manager.""" self.context_stack: list[Dict[str, Any]] = [] def push_context( self, request_id: str, request_path: str, request_method: str, user_id: Optional[str] = None, ) -> None: """ Push request context onto stack. Args: request_id: Unique request identifier request_path: Request path request_method: HTTP method user_id: User ID if available """ context = { "request_id": request_id, "request_path": request_path, "request_method": request_method, "user_id": user_id, "timestamp": datetime.now(timezone.utc).isoformat(), } self.context_stack.append(context) def pop_context(self) -> Optional[Dict[str, Any]]: """ Pop request context from stack. Returns: Context dictionary or None if empty """ return self.context_stack.pop() if self.context_stack else None def get_current_context(self) -> Optional[Dict[str, Any]]: """ Get current request context. Returns: Current context or None if empty """ return self.context_stack[-1] if self.context_stack else None # Global request context manager _context_manager: Optional[RequestContextManager] = None def get_context_manager() -> RequestContextManager: """ Get or create global context manager instance. Returns: RequestContextManager instance """ global _context_manager if _context_manager is None: _context_manager = RequestContextManager() return _context_manager