fix test and add doc

src/server/utils/error_tracking.py

@@ -0,0 +1,227 @@
+"""
+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
+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.utcnow().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.utcnow().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