from enum import Enum from typing import Optional, Dict, Any import traceback import logging import json import re from datetime import datetime logger = logging.getLogger(__name__) import re class BaseOurcoachException(Exception): def __init__(self, **kwargs): self.user_id = kwargs.get('user_id', 'no-user') self.code = kwargs.get('code', 'UnknownError') self.message = kwargs.get('message', 'An unknown error occurred') self.e = kwargs.get('e', None) # Initialize the parent Exception with the message Exception.__init__(self, self.message) # Capture the full traceback self.stack_trace = traceback.format_exc() self.timestamp = datetime.utcnow() logger.exception(f"Error raised with code={self.code}, message={self.message}, user_id={self.user_id}", exc_info=self.e) def get_formatted_details(self) -> dict: def format_traceback(traceback_str: str) -> str: # Extract error type and message error_pattern = r"(\w+Error): (.+?)(?=\n|$)" error_match = re.search(error_pattern, traceback_str) error_type = error_match.group(1) if error_match else "Unknown Error" error_msg = error_match.group(2) if error_match else "" # Extract file paths and line numbers file_pattern = r"File \"(.+?)\", line (\d+), in (\w+)" matches = re.findall(file_pattern, traceback_str) # Build formatted output formatted_lines = [f"Error: {error_type} - {error_msg}\n"] for filepath, line_num, func_name in matches: if func_name == "wrapper": continue # Convert to relative path rel_path = filepath.split('app/')[-1] if 'app/' in filepath else filepath.split('\\')[-1] formatted_lines.append(f"at {rel_path}:{func_name} (line {line_num})") return "\n".join(formatted_lines) """Returns pinpointed error details.""" return { "type": f"{self.__class__.__name__}{'.' + self.code if self.code != self.__class__.__name__ else ''}", "message": self.message, "stack_trace": format_traceback(self.stack_trace), "user_id": self.user_id, "at": self.timestamp.isoformat(), } def to_json(self) -> Dict[str, Any]: """Convert exception to JSON-serializable dictionary""" return self.get_formatted_details() def __str__(self) -> str: return json.dumps(self.to_json(), indent=2) class DBError(BaseOurcoachException): ALLOWED_CODES = ['S3Error', 'SQLError', 'NoOnboardingError', 'NoPickleError', 'NoBookingError'] def __init__(self, **kwargs): if kwargs.get('code') not in self.ALLOWED_CODES: raise ValueError(f"Invalid code for DBError: {kwargs.get('code')}") super().__init__(**kwargs) def to_json(self) -> Dict[str, Any]: base_json = super().to_json() base_json["allowed_codes"] = self.ALLOWED_CODES return base_json class OpenAIRequestError(BaseOurcoachException): def __init__(self, **kwargs): super().__init__(**kwargs) self.run_id = kwargs.get('run_id', None) def to_json(self) -> Dict[str, Any]: base_json = super().to_json() base_json["run_id"] = self.run_id return base_json class AssistantError(BaseOurcoachException): def __init__(self, **kwargs): kwargs['code'] = "AssistantError" super().__init__(**kwargs) class UserError(BaseOurcoachException): def __init__(self, **kwargs): kwargs['code'] = "UserError" super().__init__(**kwargs) class ConversationManagerError(BaseOurcoachException): def __init__(self, **kwargs): kwargs['code'] = "ConversationManagerError" super().__init__(**kwargs) class FastAPIError(BaseOurcoachException): def __init__(self, **kwargs): kwargs['code'] = "FastAPIError" super().__init__(**kwargs) class UtilsError(BaseOurcoachException): def __init__(self, **kwargs): kwargs['code'] = "UtilsError" super().__init__(**kwargs)