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