| """
|
| Custom Exceptions and Error Handling
|
| Issues #13, #14: Specific exception types, proper error logging
|
| """
|
|
|
| import logging
|
| from typing import Optional, Dict, Any
|
| from fastapi import HTTPException
|
| from datetime import datetime
|
| import uuid
|
|
|
| logger = logging.getLogger("tb_guard_errors")
|
|
|
|
|
| class TBGuardException(Exception):
|
| """Base exception for TB-Guard-XAI"""
|
|
|
| def __init__(
|
| self,
|
| message: str,
|
| code: str,
|
| status_code: int = 500,
|
| details: Optional[Dict[str, Any]] = None
|
| ):
|
| self.message = message
|
| self.code = code
|
| self.status_code = status_code
|
| self.details = details or {}
|
| self.error_id = str(uuid.uuid4())[:8]
|
| self.timestamp = datetime.utcnow().isoformat()
|
| super().__init__(message)
|
|
|
| def to_http_exception(self) -> HTTPException:
|
| """Convert to FastAPI HTTPException"""
|
| return HTTPException(
|
| status_code=self.status_code,
|
| detail={
|
| "error": self.message,
|
| "code": self.code,
|
| "error_id": self.error_id,
|
| "timestamp": self.timestamp
|
| }
|
| )
|
|
|
| def log(self, exc_info=False):
|
| """Log the exception"""
|
| logger.error(
|
| f"[{self.code}] {self.message}",
|
| extra={"error_id": self.error_id, "details": self.details},
|
| exc_info=exc_info
|
| )
|
|
|
|
|
|
|
| class InvalidImageError(TBGuardException):
|
| """Image validation failed"""
|
| def __init__(self, reason: str, details: Optional[Dict] = None):
|
| super().__init__(
|
| message=f"Invalid image: {reason}",
|
| code="INVALID_IMAGE",
|
| status_code=400,
|
| details=details
|
| )
|
|
|
|
|
| class InvalidFileError(TBGuardException):
|
| """File upload validation failed"""
|
| def __init__(self, reason: str):
|
| super().__init__(
|
| message=f"Invalid file: {reason}",
|
| code="INVALID_FILE",
|
| status_code=400
|
| )
|
|
|
|
|
| class FileTooLargeError(TBGuardException):
|
| """File exceeds size limit"""
|
| def __init__(self, file_size_mb: float, max_size_mb: int):
|
| super().__init__(
|
| message=f"File size {file_size_mb:.1f} MB exceeds limit of {max_size_mb} MB",
|
| code="FILE_TOO_LARGE",
|
| status_code=400,
|
| details={"file_size_mb": file_size_mb, "max_size_mb": max_size_mb}
|
| )
|
|
|
|
|
| class InvalidInputError(TBGuardException):
|
| """Input validation failed (symptoms, query, etc.)"""
|
| def __init__(self, field: str, reason: str):
|
| super().__init__(
|
| message=f"Invalid {field}: {reason}",
|
| code="INVALID_INPUT",
|
| status_code=400,
|
| details={"field": field}
|
| )
|
|
|
|
|
|
|
| class ModelError(TBGuardException):
|
| """Model loading or inference error"""
|
| def __init__(self, reason: str):
|
| super().__init__(
|
| message=f"Model error: {reason}",
|
| code="MODEL_ERROR",
|
| status_code=500
|
| )
|
|
|
|
|
| class ModelNotLoadedError(TBGuardException):
|
| """Model hasn't been loaded"""
|
| def __init__(self):
|
| super().__init__(
|
| message="Model is not loaded. Server startup incomplete.",
|
| code="MODEL_NOT_LOADED",
|
| status_code=503
|
| )
|
|
|
|
|
| class InferenceError(TBGuardException):
|
| """Inference (prediction) failed"""
|
| def __init__(self, reason: str):
|
| super().__init__(
|
| message=f"Inference failed: {reason}",
|
| code="INFERENCE_ERROR",
|
| status_code=500
|
| )
|
|
|
|
|
| class OutOfDistributionError(TBGuardException):
|
| """Image is out-of-distribution (anomalous)"""
|
| def __init__(self, reason: str):
|
| super().__init__(
|
| message=f"Image appears to be out-of-distribution: {reason}",
|
| code="OUT_OF_DISTRIBUTION",
|
| status_code=400
|
| )
|
|
|
|
|
|
|
| class LLMError(TBGuardException):
|
| """LLM (Mistral) API error"""
|
| def __init__(self, llm_name: str, reason: str):
|
| super().__init__(
|
| message=f"{llm_name} API error: {reason}",
|
| code=f"{llm_name.upper()}_ERROR",
|
| status_code=502
|
| )
|
|
|
|
|
| class AudioTranscriptionError(TBGuardException):
|
| """Audio transcription failed"""
|
| def __init__(self, reason: str):
|
| super().__init__(
|
| message=f"Audio transcription failed: {reason}",
|
| code="TRANSCRIPTION_ERROR",
|
| status_code=500
|
| )
|
|
|
|
|
| class RAGError(TBGuardException):
|
| """Vector database (Qdrant) error"""
|
| def __init__(self, reason: str):
|
| super().__init__(
|
| message=f"Evidence retrieval failed: {reason}",
|
| code="RAG_ERROR",
|
| status_code=503
|
| )
|
|
|
|
|
| class InternetConnectivityError(TBGuardException):
|
| """No internet connection (for cloud APIs)"""
|
| def __init__(self):
|
| super().__init__(
|
| message="No internet connection. Online mode unavailable.",
|
| code="NO_INTERNET",
|
| status_code=503
|
| )
|
|
|
|
|
|
|
| class InvalidAPIKeyError(TBGuardException):
|
| """API key validation failed"""
|
| def __init__(self, reason: str = "Invalid or expired API key"):
|
| super().__init__(
|
| message=reason,
|
| code="INVALID_API_KEY",
|
| status_code=401
|
| )
|
|
|
|
|
| class QuotaExceededError(TBGuardException):
|
| """API quota exceeded"""
|
| def __init__(self, quota_type: str = "requests"):
|
| super().__init__(
|
| message=f"Daily {quota_type} quota exceeded",
|
| code="QUOTA_EXCEEDED",
|
| status_code=429
|
| )
|
|
|
|
|
| class RateLimitError(TBGuardException):
|
| """Rate limit exceeded"""
|
| def __init__(self, reset_seconds: int):
|
| super().__init__(
|
| message=f"Rate limit exceeded. Try again in {reset_seconds} seconds.",
|
| code="RATE_LIMITED",
|
| status_code=429,
|
| details={"retry_after_seconds": reset_seconds}
|
| )
|
|
|
|
|
|
|
| class PreprocessingError(TBGuardException):
|
| """Image preprocessing failed"""
|
| def __init__(self, reason: str):
|
| super().__init__(
|
| message=f"Preprocessing failed: {reason}",
|
| code="PREPROCESSING_ERROR",
|
| status_code=400
|
| )
|
|
|
|
|
| class CorruptedImageError(TBGuardException):
|
| """Image appears corrupted"""
|
| def __init__(self, reason: str):
|
| super().__init__(
|
| message=f"Image appears corrupted: {reason}",
|
| code="CORRUPTED_IMAGE",
|
| status_code=400
|
| )
|
|
|
|
|
|
|
| class ConfigurationError(TBGuardException):
|
| """Configuration validation failed (startup)"""
|
| def __init__(self, reason: str):
|
| super().__init__(
|
| message=f"Configuration error: {reason}",
|
| code="CONFIG_ERROR",
|
| status_code=500
|
| )
|
|
|
|
|
| class MissingEnvironmentVariableError(ConfigurationError):
|
| """Required environment variable not set"""
|
| def __init__(self, var_name: str):
|
| super().__init__(f"Missing required environment variable: {var_name}")
|
|
|
|
|
|
|
| class PerformanceDegradationError(TBGuardException):
|
| """Model performance degraded (drift detected)"""
|
| def __init__(self, accuracy_drop: float):
|
| super().__init__(
|
| message=f"Model performance degraded by {accuracy_drop:.1%}. Manual review required.",
|
| code="PERFORMANCE_DEGRADATION",
|
| status_code=503
|
| )
|
|
|
|
|
|
|
| def handle_exception(exc: Exception) -> HTTPException:
|
| """
|
| Convert any exception to appropriate HTTP response
|
| """
|
| if isinstance(exc, TBGuardException):
|
| exc.log(exc_info=True)
|
| return exc.to_http_exception()
|
|
|
|
|
| error_id = str(uuid.uuid4())[:8]
|
| logger.exception(f"Unexpected error [{error_id}]", extra={"error_id": error_id})
|
|
|
| return HTTPException(
|
| status_code=500,
|
| detail={
|
| "error": "Internal server error",
|
| "code": "INTERNAL_ERROR",
|
| "error_id": error_id,
|
| "message": "An unexpected error occurred. Please contact support with the error ID."
|
| }
|
| )
|
|
|