fastapi-v2 / app /exceptions.py
BMCVRN's picture
Error Handling (#7)
0c4649e verified
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)