| import logging
|
| import traceback
|
| from datetime import datetime
|
| import json
|
| from pathlib import Path
|
| from typing import Dict, Any, Optional
|
|
|
|
|
| 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 {}
|
| }
|
|
|
|
|
| self.errors[error_id] = error_data
|
|
|
|
|
| self._save_errors()
|
|
|
|
|
| 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"]
|
|
|
|
|
| for error_type in summary:
|
| summary[error_type]["endpoints"] = list(summary[error_type]["endpoints"])
|
|
|
| return summary |
