29e42b8d80
This commit introduces significant updates to the cost management functionality, including: - New endpoints for retrieving cost summaries and generating expenses from lists, improving user interaction with financial data. - Implementation of a service layer for cost-related logic, encapsulating the core business rules for calculating cost summaries and managing expenses. - Introduction of a financial activity endpoint to consolidate user expenses and settlements, enhancing the user experience by providing a comprehensive view of financial activities. - Refactoring of existing code to improve maintainability and clarity, including the addition of new exception handling for financial conflicts. These changes aim to streamline cost management processes and enhance the overall functionality of the application.
373 lines
13 KiB
Python
373 lines
13 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 = "An error occurred during a settlement operation."):
|
|
super().__init__(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=detail
|
|
)
|
|
|
|
class FinancialConflictError(HTTPException):
|
|
"""Raised when a financial conflict occurs."""
|
|
def __init__(self, detail: str):
|
|
super().__init__(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail=detail
|
|
)
|
|
|
|
class ConflictError(HTTPException):
|
|
"""Raised when a 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.AUTH_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.AUTH_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
|
|
)
|
|
|
|
class UserOperationError(HTTPException):
|
|
"""Raised when a user operation fails."""
|
|
def __init__(self, detail: str):
|
|
super().__init__(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=detail
|
|
)
|
|
|
|
class ChoreOperationError(HTTPException):
|
|
"""Raised when a chore-related operation fails."""
|
|
def __init__(self, detail: str):
|
|
super().__init__(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=detail
|
|
)
|
|
|
|
class ChoreNotFoundError(HTTPException):
|
|
"""Raised when a chore or assignment is not found."""
|
|
def __init__(self, chore_id: int = None, assignment_id: int = None, group_id: Optional[int] = None, detail: Optional[str] = None):
|
|
if detail:
|
|
error_detail = detail
|
|
elif group_id is not None:
|
|
error_detail = f"Chore {chore_id} not found in group {group_id}"
|
|
else:
|
|
error_detail = f"Chore {chore_id} not found"
|
|
super().__init__(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=error_detail
|
|
)
|
|
|
|
class PermissionDeniedError(HTTPException):
|
|
"""Raised when a user is denied permission for an action."""
|
|
def __init__(self, detail: str = "Permission denied."):
|
|
super().__init__(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=detail
|
|
)
|
|
|