162 lines
4.5 KiB
Python
162 lines
4.5 KiB
Python
"""
|
|
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 {}}
|
|
)
|