| from collections.abc import Callable | |||||
| from datetime import UTC, datetime | from datetime import UTC, datetime | ||||
| from typing import Optional, Union | from typing import Optional, Union | ||||
| return InfiniteScrollPagination(data=conversations, limit=limit, has_more=has_more) | return InfiniteScrollPagination(data=conversations, limit=limit, has_more=has_more) | ||||
| @classmethod | @classmethod | ||||
| def _get_sort_params(cls, sort_by: str) -> tuple[str, callable]: | |||||
| def _get_sort_params(cls, sort_by: str): | |||||
| if sort_by.startswith("-"): | if sort_by.startswith("-"): | ||||
| return sort_by[1:], desc | return sort_by[1:], desc | ||||
| return sort_by, asc | return sort_by, asc | ||||
| @classmethod | @classmethod | ||||
| def _build_filter_condition( | def _build_filter_condition( | ||||
| cls, sort_field: str, sort_direction: callable, reference_conversation: Conversation, is_next_page: bool = False | |||||
| cls, sort_field: str, sort_direction: Callable, reference_conversation: Conversation, is_next_page: bool = False | |||||
| ): | ): | ||||
| field_value = getattr(reference_conversation, sort_field) | field_value = getattr(reference_conversation, sort_field) | ||||
| if (sort_direction == desc and not is_next_page) or (sort_direction == asc and is_next_page): | if (sort_direction == desc and not is_next_page) or (sort_direction == asc and is_next_page): | ||||
| conversation = cls.get_conversation(app_model, conversation_id, user) | conversation = cls.get_conversation(app_model, conversation_id, user) | ||||
| conversation.is_deleted = True | conversation.is_deleted = True | ||||
| conversation.updated_at = datetime.now(timezone.utc).replace(tzinfo=None) | |||||
| conversation.updated_at = datetime.now(UTC).replace(tzinfo=None) | |||||
| db.session.commit() | db.session.commit() |