Upload error_logger.py

error_logger.py

@@ -0,0 +1,122 @@
+import logging
+import traceback
+from datetime import datetime
+import json
+from pathlib import Path
+from typing import Dict, Any, Optional
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+    handlers=[
+        logging.FileHandler('api_errors.log'),
+        logging.StreamHandler()
+    ]
+)
+
+logger = logging.getLogger("api.errors")
+
+class ErrorLogger:
+    def __init__(self):
+        self.error_log_file = "detailed_errors.json"
+        self.errors: Dict[str, Any] = self._load_errors()
+
+    def _load_errors(self) -> Dict[str, Any]:
+        """Load existing errors from file"""
+        try:
+            if Path(self.error_log_file).exists():
+                with open(self.error_log_file, 'r') as f:
+                    return json.load(f)
+        except Exception as e:
+            logger.error(f"Failed to load error log: {e}")
+        return {}
+
+    def _save_errors(self):
+        """Save errors to file"""
+        try:
+            with open(self.error_log_file, 'w') as f:
+                json.dump(self.errors, f, indent=2)
+        except Exception as e:
+            logger.error(f"Failed to save error log: {e}")
+
+    def log_error(self, 
+                  error: Exception,
+                  endpoint: str,
+                  request_info: Optional[Dict] = None,
+                  context: Optional[Dict] = None):
+        """
+        Log detailed error information
+        
+        Args:
+            error: The exception that occurred
+            endpoint: The API endpoint where the error occurred
+            request_info: Dictionary containing request information
+            context: Additional context about the error
+        """
+        timestamp = datetime.now().isoformat()
+        error_id = f"{timestamp}-{hash(str(error))}"
+
+        error_data = {
+            "timestamp": timestamp,
+            "error_type": error.__class__.__name__,
+            "error_message": str(error),
+            "endpoint": endpoint,
+            "traceback": traceback.format_exc(),
+            "request_info": request_info or {},
+            "context": context or {}
+        }
+
+        # Store in memory
+        self.errors[error_id] = error_data
+        
+        # Save to file
+        self._save_errors()
+
+        # Log to standard logging
+        logger.error(
+            f"Error in {endpoint}: {error.__class__.__name__} - {str(error)}\n"
+            f"Context: {json.dumps(context or {}, indent=2)}\n"
+            f"Request: {json.dumps(request_info or {}, indent=2)}\n"
+            f"Traceback: {traceback.format_exc()}"
+        )
+
+        return error_id
+
+    def get_error(self, error_id: str) -> Optional[Dict]:
+        """Retrieve detailed error information by ID"""
+        return self.errors.get(error_id)
+
+    def get_recent_errors(self, limit: int = 10) -> Dict[str, Any]:
+        """Get most recent errors"""
+        sorted_errors = sorted(
+            self.errors.items(),
+            key=lambda x: x[1]["timestamp"],
+            reverse=True
+        )
+        return dict(sorted_errors[:limit])
+
+    def get_error_summary(self) -> Dict[str, Any]:
+        """Get summary of errors by type"""
+        summary = {}
+        for error_id, error_data in self.errors.items():
+            error_type = error_data["error_type"]
+            if error_type not in summary:
+                summary[error_type] = {
+                    "count": 0,
+                    "last_occurrence": None,
+                    "endpoints": set()
+                }
+            
+            summary[error_type]["count"] += 1
+            summary[error_type]["endpoints"].add(error_data["endpoint"])
+            
+            if (not summary[error_type]["last_occurrence"] or 
+                error_data["timestamp"] > summary[error_type]["last_occurrence"]):
+                summary[error_type]["last_occurrence"] = error_data["timestamp"]
+        
+        # Convert sets to lists for JSON serialization
+        for error_type in summary:
+            summary[error_type]["endpoints"] = list(summary[error_type]["endpoints"])
+            
+        return summary