Spaces:
Sleeping
Sleeping
File size: 4,243 Bytes
0c4649e | 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 | 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)
|