|
|
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 |
