130 lines
6.8 KiB
Python
130 lines
6.8 KiB
Python
from loguru import logger
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from .business_exception import BusinessValidationException
|
|
from .resource_not_found_exception import ResourceNotFoundException
|
|
from .validation_exception import ValidationException
|
|
|
|
class DBExceptionHandler:
|
|
"""
|
|
Centralized handler for database exceptions.
|
|
This class provides methods to handle and transform database exceptions
|
|
into application-specific exceptions with user-friendly messages.
|
|
"""
|
|
@staticmethod
|
|
def _extract_detail_message(e):
|
|
"""
|
|
Extract the detailed error message from a database exception.
|
|
|
|
Args:
|
|
e: The exception to extract the message from
|
|
|
|
Returns:
|
|
str: The detailed error message if found, None otherwise
|
|
"""
|
|
if hasattr(e, 'args') and e.args and '\nDETAIL:' in str(e.args[0]):
|
|
# Extract just the part after 'DETAIL:'
|
|
detailed_message = str(e.args[0]).split('\nDETAIL:')[1].strip()
|
|
# Clean up any trailing newlines or other characters
|
|
detailed_message = detailed_message.split('\n')[0].strip()
|
|
return detailed_message
|
|
return None
|
|
|
|
@staticmethod
|
|
def handle_exception(e, context="database operation"):
|
|
"""
|
|
Handle database exceptions and convert them to application-specific exceptions.
|
|
|
|
Args:
|
|
e: The exception to handle
|
|
context: A string describing the context of the operation (for logging)
|
|
|
|
Raises:
|
|
BusinessValidationException: With a user-friendly message
|
|
ValidationException: If the original exception was a ValidationException
|
|
ResourceNotFoundException: If the original exception was a ResourceNotFoundException
|
|
"""
|
|
logger.error(f"Error during {context}: {str(e)}")
|
|
|
|
# Pass through our custom exceptions
|
|
if isinstance(e, ValidationException):
|
|
raise ValidationException(e.message)
|
|
if isinstance(e, ResourceNotFoundException):
|
|
raise ResourceNotFoundException(e.message)
|
|
if isinstance(e, BusinessValidationException):
|
|
raise BusinessValidationException(e.message)
|
|
|
|
# Handle SQLAlchemy errors
|
|
if isinstance(e, SQLAlchemyError):
|
|
# Check for PostgreSQL unique constraint violations
|
|
if hasattr(e, '__cause__') and hasattr(e.__cause__, '__class__') and e.__cause__.__class__.__name__ == 'UniqueViolation':
|
|
# Try to extract the detailed error message
|
|
detailed_message = DBExceptionHandler._extract_detail_message(e)
|
|
if detailed_message:
|
|
raise BusinessValidationException(detailed_message)
|
|
|
|
# Fallback to extracting field from the error if we couldn't get the detailed message
|
|
field = None
|
|
if 'Key (' in str(e.__cause__) and ')=' in str(e.__cause__):
|
|
field = str(e.__cause__).split('Key (')[1].split(')=')[0]
|
|
|
|
# Generic message if we couldn't extract a better one
|
|
raise BusinessValidationException(f"A record with this {field or 'information'} already exists")
|
|
|
|
# Handle foreign key constraint violations
|
|
elif hasattr(e, '__cause__') and hasattr(e.__cause__, '__class__') and e.__cause__.__class__.__name__ == 'ForeignKeyViolation':
|
|
# Try to extract detailed message
|
|
detailed_message = DBExceptionHandler._extract_detail_message(e)
|
|
if detailed_message:
|
|
raise BusinessValidationException(detailed_message)
|
|
raise BusinessValidationException(f"Referenced record does not exist")
|
|
|
|
# Handle check constraint violations
|
|
elif hasattr(e, '__cause__') and hasattr(e.__cause__, '__class__') and e.__cause__.__class__.__name__ == 'CheckViolation':
|
|
# Try to extract detailed message
|
|
detailed_message = DBExceptionHandler._extract_detail_message(e)
|
|
if detailed_message:
|
|
raise BusinessValidationException(detailed_message)
|
|
raise BusinessValidationException(f"Invalid data: failed validation check")
|
|
|
|
# Handle not null constraint violations
|
|
elif hasattr(e, '__cause__') and hasattr(e.__cause__, '__class__') and e.__cause__.__class__.__name__ == 'NotNullViolation':
|
|
# Try to extract detailed message
|
|
detailed_message = DBExceptionHandler._extract_detail_message(e)
|
|
if detailed_message:
|
|
raise BusinessValidationException(detailed_message)
|
|
|
|
# Fallback to extracting field name
|
|
field = None
|
|
if 'column "' in str(e.__cause__) and '" violates' in str(e.__cause__):
|
|
field = str(e.__cause__).split('column "')[1].split('" violates')[0]
|
|
raise BusinessValidationException(f"Required field {field or ''} cannot be empty")
|
|
|
|
# Generic SQLAlchemy error
|
|
elif hasattr(e, '__cause__') and hasattr(e.__cause__, '__class__') and e.__cause__.__class__.__name__ == 'IntegrityError':
|
|
# Try to extract detailed message
|
|
detailed_message = DBExceptionHandler._extract_detail_message(e)
|
|
if detailed_message:
|
|
raise BusinessValidationException(detailed_message)
|
|
raise BusinessValidationException(f"Database error: {e.__class__.__name__}")
|
|
|
|
# Handle unique constraint violations (redundant with first check, but keeping for safety)
|
|
elif hasattr(e, '__cause__') and hasattr(e.__cause__, '__class__') and e.__cause__.__class__.__name__ == 'UniqueViolation':
|
|
# Try to extract detailed message
|
|
detailed_message = DBExceptionHandler._extract_detail_message(e)
|
|
if detailed_message:
|
|
raise BusinessValidationException(detailed_message)
|
|
raise BusinessValidationException(f"A record with this information already exists")
|
|
|
|
# Handle other database errors
|
|
elif hasattr(e, '__cause__') and hasattr(e.__cause__, '__class__') and e.__cause__.__class__.__name__ == 'OperationalError':
|
|
# Try to extract detailed message
|
|
detailed_message = DBExceptionHandler._extract_detail_message(e)
|
|
if detailed_message:
|
|
raise BusinessValidationException(detailed_message)
|
|
raise BusinessValidationException(f"Database error: {e.__class__.__name__}")
|
|
|
|
|
|
|
|
# For any other exceptions, provide a generic message
|
|
raise BusinessValidationException(f"An error occurred: {e.__class__.__name__}")
|