fastapi-v2 / app /exceptions.py
Shageenderan Sapai
[FEATURE] Error Handling
f5fccf5
raw
history blame
4.24 kB
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)