""" Custom exceptions for MongoDB operations and password management. This module defines custom exceptions for handling various error cases in MongoDB operations and password-related functionality. """ from typing import Any, Dict, Optional from fastapi import HTTPException, status from ErrorHandlers.ErrorHandlers.api_exc_handler import HTTPExceptionApi class MongoBaseException(Exception): """Base exception for MongoDB-related errors.""" def __init__( self, message: str, status_code: int = status.HTTP_500_INTERNAL_SERVER_ERROR, details: Optional[Dict[str, Any]] = None, ): self.message = message self.status_code = status_code self.details = details or {} super().__init__(self.message) def to_http_exception(self) -> HTTPException: """Convert to FastAPI HTTPException.""" return HTTPExceptionApi( lang="en", error_code=self.status_code, ) class MongoConnectionError(MongoBaseException): """Raised when there's an error connecting to MongoDB.""" def __init__( self, message: str = "Failed to connect to MongoDB", details: Optional[Dict[str, Any]] = None, ): super().__init__( message=message, status_code=status.HTTP_503_SERVICE_UNAVAILABLE, details=details, ) class MongoDocumentNotFoundError(MongoBaseException): """Raised when a document is not found in MongoDB.""" def __init__( self, collection: str, filter_query: Dict[str, Any], message: Optional[str] = None, ): message = message or f"Document not found in collection '{collection}'" super().__init__( message=message, status_code=status.HTTP_404_NOT_FOUND, details={"collection": collection, "filter": filter_query}, ) class MongoValidationError(MongoBaseException): """Raised when document validation fails.""" def __init__(self, message: str, field_errors: Optional[Dict[str, str]] = None): super().__init__( message=message, status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, details={"field_errors": field_errors or {}}, ) class MongoDuplicateKeyError(MongoBaseException): """Raised when trying to insert a document with a duplicate key.""" def __init__( self, collection: str, key_pattern: Dict[str, Any], message: Optional[str] = None, ): message = message or f"Duplicate key error in collection '{collection}'" super().__init__( message=message, status_code=status.HTTP_409_CONFLICT, details={"collection": collection, "key_pattern": key_pattern}, ) class PasswordHistoryError(MongoBaseException): """Base exception for password history-related errors.""" def __init__( self, message: str, status_code: int = status.HTTP_400_BAD_REQUEST, details: Optional[Dict[str, Any]] = None, ): super().__init__(message, status_code, details) class PasswordReuseError(PasswordHistoryError): """Raised when attempting to reuse a recent password.""" def __init__( self, message: str = "Password was used recently", history_limit: Optional[int] = None, ): details = {"history_limit": history_limit} if history_limit else None super().__init__( message=message, status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, details=details, ) class PasswordHistoryLimitError(PasswordHistoryError): """Raised when password history limit is reached.""" def __init__(self, limit: int, message: Optional[str] = None): message = message or f"Password history limit of {limit} reached" super().__init__( message=message, status_code=status.HTTP_409_CONFLICT, details={"limit": limit}, ) class InvalidPasswordDetailError(PasswordHistoryError): """Raised when password history detail is invalid.""" def __init__(self, message: str, field_errors: Optional[Dict[str, str]] = None): super().__init__( message=message, status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, details={"field_errors": field_errors or {}}, )