Underground-Digital/Workflow-Engine

1from typing import Optional, Union2 3from extensions.ext_database import db4from libs.infinite_scroll_pagination import InfiniteScrollPagination5from models.account import Account6from models.model import App, EndUser7from models.web import SavedMessage8from services.message_service import MessageService9 10 11class SavedMessageService:12    @classmethod13    def pagination_by_last_id(14        cls, app_model: App, user: Optional[Union[Account, EndUser]], last_id: Optional[str], limit: int15    ) -> InfiniteScrollPagination:16        saved_messages = (17            db.session.query(SavedMessage)18            .filter(19                SavedMessage.app_id == app_model.id,20                SavedMessage.created_by_role == ("account" if isinstance(user, Account) else "end_user"),21                SavedMessage.created_by == user.id,22            )23            .order_by(SavedMessage.created_at.desc())24            .all()25        )26        message_ids = [sm.message_id for sm in saved_messages]27 28        return MessageService.pagination_by_last_id(29            app_model=app_model, user=user, last_id=last_id, limit=limit, include_ids=message_ids30        )31 32    @classmethod33    def save(cls, app_model: App, user: Optional[Union[Account, EndUser]], message_id: str):34        saved_message = (35            db.session.query(SavedMessage)36            .filter(37                SavedMessage.app_id == app_model.id,38                SavedMessage.message_id == message_id,39                SavedMessage.created_by_role == ("account" if isinstance(user, Account) else "end_user"),40                SavedMessage.created_by == user.id,41            )42            .first()43        )44 45        if saved_message:46            return47 48        message = MessageService.get_message(app_model=app_model, user=user, message_id=message_id)49 50        saved_message = SavedMessage(51            app_id=app_model.id,52            message_id=message.id,53            created_by_role="account" if isinstance(user, Account) else "end_user",54            created_by=user.id,55        )56 57        db.session.add(saved_message)58        db.session.commit()59 60    @classmethod61    def delete(cls, app_model: App, user: Optional[Union[Account, EndUser]], message_id: str):62        saved_message = (63            db.session.query(SavedMessage)64            .filter(65                SavedMessage.app_id == app_model.id,66                SavedMessage.message_id == message_id,67                SavedMessage.created_by_role == ("account" if isinstance(user, Account) else "end_user"),68                SavedMessage.created_by == user.id,69            )70            .first()71        )72 73        if not saved_message:74            return75 76        db.session.delete(saved_message)77        db.session.commit()78