Files
orion/app/exceptions/message.py
Samir Boulahtit 0098093287 feat: enhance messaging system with improved API and tests
- Refactor messaging API endpoints for admin, shop, and vendor
- Add message-specific exceptions (ConversationNotFoundException, etc.)
- Enhance messaging service with additional helper methods
- Add comprehensive test fixtures for messaging
- Add integration tests for admin and vendor messaging APIs
- Add unit tests for messaging and attachment services

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-21 21:01:14 +01:00

101 lines
3.4 KiB
Python

# app/exceptions/message.py
"""
Messaging specific exceptions.
"""
from .base import BusinessLogicException, ResourceNotFoundException, ValidationException
class ConversationNotFoundException(ResourceNotFoundException):
"""Raised when a conversation is not found."""
def __init__(self, conversation_identifier: str):
super().__init__(
resource_type="Conversation",
identifier=conversation_identifier,
message=f"Conversation '{conversation_identifier}' not found",
error_code="CONVERSATION_NOT_FOUND",
)
class MessageNotFoundException(ResourceNotFoundException):
"""Raised when a message is not found."""
def __init__(self, message_identifier: str):
super().__init__(
resource_type="Message",
identifier=message_identifier,
message=f"Message '{message_identifier}' not found",
error_code="MESSAGE_NOT_FOUND",
)
class ConversationClosedException(BusinessLogicException):
"""Raised when trying to send message to a closed conversation."""
def __init__(self, conversation_id: int):
super().__init__(
message=f"Cannot send message to closed conversation {conversation_id}",
error_code="CONVERSATION_CLOSED",
details={"conversation_id": conversation_id},
)
class MessageAttachmentException(ValidationException):
"""Raised when attachment validation fails."""
def __init__(self, message: str, details: dict | None = None):
super().__init__(
message=message,
error_code="MESSAGE_ATTACHMENT_INVALID",
details=details,
)
class UnauthorizedConversationAccessException(BusinessLogicException):
"""Raised when user tries to access a conversation they don't have access to."""
def __init__(self, conversation_id: int):
super().__init__(
message=f"You do not have access to conversation {conversation_id}",
error_code="CONVERSATION_ACCESS_DENIED",
details={"conversation_id": conversation_id},
)
class InvalidConversationTypeException(ValidationException):
"""Raised when conversation type is not valid for the operation."""
def __init__(self, message: str, allowed_types: list[str] | None = None):
super().__init__(
message=message,
error_code="INVALID_CONVERSATION_TYPE",
details={"allowed_types": allowed_types} if allowed_types else None,
)
class InvalidRecipientTypeException(ValidationException):
"""Raised when recipient type doesn't match conversation type."""
def __init__(self, conversation_type: str, expected_recipient_type: str):
super().__init__(
message=f"{conversation_type} conversations require a {expected_recipient_type} recipient",
error_code="INVALID_RECIPIENT_TYPE",
details={
"conversation_type": conversation_type,
"expected_recipient_type": expected_recipient_type,
},
)
class AttachmentNotFoundException(ResourceNotFoundException):
"""Raised when an attachment is not found."""
def __init__(self, attachment_id: int | str):
super().__init__(
resource_type="Attachment",
identifier=str(attachment_id),
message=f"Attachment '{attachment_id}' not found",
error_code="ATTACHMENT_NOT_FOUND",
)