325 lines
12 KiB
Python
325 lines
12 KiB
Python
from fastapi import HTTPException, status
|
|
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
|
from app.config import settings
|
|
from typing import Optional
|
|
|
|
class ListNotFoundError(HTTPException):
|
|
"""Raised when a list is not found."""
|
|
def __init__(self, list_id: int):
|
|
super().__init__(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"List {list_id} not found"
|
|
)
|
|
|
|
class ListPermissionError(HTTPException):
|
|
"""Raised when a user doesn't have permission to access a list."""
|
|
def __init__(self, list_id: int, action: str = "access"):
|
|
super().__init__(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=f"You do not have permission to {action} list {list_id}"
|
|
)
|
|
|
|
class ListCreatorRequiredError(HTTPException):
|
|
"""Raised when an action requires the list creator but the user is not the creator."""
|
|
def __init__(self, list_id: int, action: str):
|
|
super().__init__(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=f"Only the list creator can {action} list {list_id}"
|
|
)
|
|
|
|
class GroupNotFoundError(HTTPException):
|
|
"""Raised when a group is not found."""
|
|
def __init__(self, group_id: int):
|
|
super().__init__(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Group {group_id} not found"
|
|
)
|
|
|
|
class GroupPermissionError(HTTPException):
|
|
"""Raised when a user doesn't have permission to perform an action in a group."""
|
|
def __init__(self, group_id: int, action: str):
|
|
super().__init__(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=f"You do not have permission to {action} in group {group_id}"
|
|
)
|
|
|
|
class GroupMembershipError(HTTPException):
|
|
"""Raised when a user attempts to perform an action that requires group membership."""
|
|
def __init__(self, group_id: int, action: str = "access"):
|
|
super().__init__(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=f"You must be a member of group {group_id} to {action}"
|
|
)
|
|
|
|
class GroupOperationError(HTTPException):
|
|
"""Raised when a group operation fails."""
|
|
def __init__(self, detail: str):
|
|
super().__init__(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=detail
|
|
)
|
|
|
|
class GroupValidationError(HTTPException):
|
|
"""Raised when a group operation is invalid."""
|
|
def __init__(self, detail: str):
|
|
super().__init__(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=detail
|
|
)
|
|
|
|
class ItemNotFoundError(HTTPException):
|
|
"""Raised when an item is not found."""
|
|
def __init__(self, item_id: int):
|
|
super().__init__(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Item {item_id} not found"
|
|
)
|
|
|
|
class UserNotFoundError(HTTPException):
|
|
"""Raised when a user is not found."""
|
|
def __init__(self, user_id: Optional[int] = None, identifier: Optional[str] = None):
|
|
detail_msg = "User not found."
|
|
if user_id:
|
|
detail_msg = f"User with ID {user_id} not found."
|
|
elif identifier:
|
|
detail_msg = f"User with identifier '{identifier}' not found."
|
|
super().__init__(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=detail_msg
|
|
)
|
|
|
|
class InvalidOperationError(HTTPException):
|
|
"""Raised when an operation is invalid or disallowed by business logic."""
|
|
def __init__(self, detail: str, status_code: int = status.HTTP_400_BAD_REQUEST):
|
|
super().__init__(
|
|
status_code=status_code,
|
|
detail=detail
|
|
)
|
|
|
|
class DatabaseConnectionError(HTTPException):
|
|
"""Raised when there is an error connecting to the database."""
|
|
def __init__(self):
|
|
super().__init__(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail=settings.DB_CONNECTION_ERROR
|
|
)
|
|
|
|
class DatabaseIntegrityError(HTTPException):
|
|
"""Raised when a database integrity constraint is violated."""
|
|
def __init__(self):
|
|
super().__init__(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=settings.DB_INTEGRITY_ERROR
|
|
)
|
|
|
|
class DatabaseTransactionError(HTTPException):
|
|
"""Raised when a database transaction fails."""
|
|
def __init__(self, detail: str = settings.DB_TRANSACTION_ERROR):
|
|
super().__init__(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=detail
|
|
)
|
|
|
|
class DatabaseQueryError(HTTPException):
|
|
"""Raised when a database query fails."""
|
|
def __init__(self, detail: str = settings.DB_QUERY_ERROR):
|
|
super().__init__(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=detail
|
|
)
|
|
|
|
class ExpenseOperationError(HTTPException):
|
|
"""Raised when an expense operation fails."""
|
|
def __init__(self, detail: str):
|
|
super().__init__(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=detail
|
|
)
|
|
|
|
class OCRServiceUnavailableError(HTTPException):
|
|
"""Raised when the OCR service is unavailable."""
|
|
def __init__(self):
|
|
super().__init__(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail=settings.OCR_SERVICE_UNAVAILABLE
|
|
)
|
|
|
|
class OCRServiceConfigError(HTTPException):
|
|
"""Raised when there is an error in the OCR service configuration."""
|
|
def __init__(self):
|
|
super().__init__(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=settings.OCR_SERVICE_CONFIG_ERROR
|
|
)
|
|
|
|
class OCRUnexpectedError(HTTPException):
|
|
"""Raised when there is an unexpected error in the OCR service."""
|
|
def __init__(self):
|
|
super().__init__(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=settings.OCR_UNEXPECTED_ERROR
|
|
)
|
|
|
|
class OCRQuotaExceededError(HTTPException):
|
|
"""Raised when the OCR service quota is exceeded."""
|
|
def __init__(self):
|
|
super().__init__(
|
|
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
|
|
detail=settings.OCR_QUOTA_EXCEEDED
|
|
)
|
|
|
|
class InvalidFileTypeError(HTTPException):
|
|
"""Raised when an invalid file type is uploaded for OCR."""
|
|
def __init__(self):
|
|
super().__init__(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=settings.OCR_INVALID_FILE_TYPE.format(types=", ".join(settings.ALLOWED_IMAGE_TYPES))
|
|
)
|
|
|
|
class FileTooLargeError(HTTPException):
|
|
"""Raised when an uploaded file exceeds the size limit."""
|
|
def __init__(self):
|
|
super().__init__(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=settings.OCR_FILE_TOO_LARGE.format(size=settings.MAX_FILE_SIZE_MB)
|
|
)
|
|
|
|
class OCRProcessingError(HTTPException):
|
|
"""Raised when there is an error processing the image with OCR."""
|
|
def __init__(self, detail: str):
|
|
super().__init__(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=settings.OCR_PROCESSING_ERROR.format(detail=detail)
|
|
)
|
|
|
|
class EmailAlreadyRegisteredError(HTTPException):
|
|
"""Raised when attempting to register with an email that is already in use."""
|
|
def __init__(self):
|
|
super().__init__(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Email already registered."
|
|
)
|
|
|
|
class UserCreationError(HTTPException):
|
|
"""Raised when there is an error creating a new user."""
|
|
def __init__(self):
|
|
super().__init__(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="An error occurred during user creation."
|
|
)
|
|
|
|
class InviteNotFoundError(HTTPException):
|
|
"""Raised when an invite is not found."""
|
|
def __init__(self, invite_code: str):
|
|
super().__init__(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Invite code {invite_code} not found"
|
|
)
|
|
|
|
class InviteExpiredError(HTTPException):
|
|
"""Raised when an invite has expired."""
|
|
def __init__(self, invite_code: str):
|
|
super().__init__(
|
|
status_code=status.HTTP_410_GONE,
|
|
detail=f"Invite code {invite_code} has expired"
|
|
)
|
|
|
|
class InviteAlreadyUsedError(HTTPException):
|
|
"""Raised when an invite has already been used."""
|
|
def __init__(self, invite_code: str):
|
|
super().__init__(
|
|
status_code=status.HTTP_410_GONE,
|
|
detail=f"Invite code {invite_code} has already been used"
|
|
)
|
|
|
|
class InviteCreationError(HTTPException):
|
|
"""Raised when an invite cannot be created."""
|
|
def __init__(self, group_id: int):
|
|
super().__init__(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Failed to create invite for group {group_id}"
|
|
)
|
|
|
|
class ListStatusNotFoundError(HTTPException):
|
|
"""Raised when a list's status cannot be retrieved."""
|
|
def __init__(self, list_id: int):
|
|
super().__init__(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Status for list {list_id} not found"
|
|
)
|
|
|
|
class InviteOperationError(HTTPException):
|
|
"""Raised when an invite operation fails."""
|
|
def __init__(self, detail: str):
|
|
super().__init__(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=detail
|
|
)
|
|
|
|
class SettlementOperationError(HTTPException):
|
|
"""Raised when a settlement operation fails."""
|
|
def __init__(self, detail: str):
|
|
super().__init__(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=detail
|
|
)
|
|
|
|
class ConflictError(HTTPException):
|
|
"""Raised when an optimistic lock version conflict occurs."""
|
|
def __init__(self, detail: str):
|
|
super().__init__(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail=detail
|
|
)
|
|
|
|
class InvalidCredentialsError(HTTPException):
|
|
"""Raised when login credentials are invalid."""
|
|
def __init__(self):
|
|
super().__init__(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=settings.AUTH_INVALID_CREDENTIALS,
|
|
headers={settings.AUTH_HEADER_NAME: f"{settings.AUTH_HEADER_PREFIX} error=\"invalid_credentials\""}
|
|
)
|
|
|
|
class NotAuthenticatedError(HTTPException):
|
|
"""Raised when the user is not authenticated."""
|
|
def __init__(self):
|
|
super().__init__(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=settings.AUTH_NOT_AUTHENTICATED,
|
|
headers={settings.AUTH_HEADER_NAME: f"{settings.AUTH_HEADER_PREFIX} error=\"not_authenticated\""}
|
|
)
|
|
|
|
class JWTError(HTTPException):
|
|
"""Raised when there is an error with the JWT token."""
|
|
def __init__(self, error: str):
|
|
super().__init__(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=settings.JWT_ERROR.format(error=error),
|
|
headers={settings.AUTH_HEADER_NAME: f"{settings.AUTH_HEADER_PREFIX} error=\"invalid_token\""}
|
|
)
|
|
|
|
class JWTUnexpectedError(HTTPException):
|
|
"""Raised when there is an unexpected error with the JWT token."""
|
|
def __init__(self, error: str):
|
|
super().__init__(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=settings.JWT_UNEXPECTED_ERROR.format(error=error),
|
|
headers={settings.AUTH_HEADER_NAME: f"{settings.AUTH_HEADER_PREFIX} error=\"invalid_token\""}
|
|
)
|
|
|
|
class ListOperationError(HTTPException):
|
|
"""Raised when a list operation fails."""
|
|
def __init__(self, detail: str):
|
|
super().__init__(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=detail
|
|
)
|
|
|
|
class ItemOperationError(HTTPException):
|
|
"""Raised when an item operation fails."""
|
|
def __init__(self, detail: str):
|
|
super().__init__(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=detail
|
|
) |