import sqlite3 import json import uuid import traceback from datetime import datetime from typing import Dict, Any, Optional class ErrorLogger: def __init__(self, db_path: str = "./errors.db"): self.db_path = db_path self.setup_db() def setup_db(self) -> None: with sqlite3.connect(self.db_path) as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS errors ( id TEXT PRIMARY KEY, timestamp DATETIME, provider TEXT, request_method TEXT, request_params TEXT, error_type TEXT, error_message TEXT, error_traceback TEXT ) """) conn.commit() def log_error(self, provider: str, request: Dict[str, Any], error: Exception) -> str: error_id = str(uuid.uuid4()) timestamp = datetime.now() with sqlite3.connect(self.db_path) as conn: conn.execute(""" INSERT INTO errors ( id, timestamp, provider, request_method, request_params, error_type, error_message, error_traceback ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( error_id, timestamp, provider, request.get("method", "unknown"), json.dumps(request.get("params", {})), type(error).__name__, str(error), traceback.format_exc() )) conn.commit() return error_id def get_error(self, error_id: str) -> Optional[Dict[str, Any]]: with sqlite3.connect(self.db_path) as conn: conn.row_factory = sqlite3.Row cursor = conn.execute( "SELECT * FROM errors WHERE id = ?", (error_id,) ) row = cursor.fetchone() if row: return { "id": row["id"], "timestamp": row["timestamp"], "provider": row["provider"], "request_method": row["request_method"], "request_params": json.loads(row["request_params"]), "error_type": row["error_type"], "error_message": row["error_message"], "error_traceback": row["error_traceback"] } return None