File size: 4,246 Bytes
36fedea |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 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 |