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)