Merge pull request #653 from Studio-Yandex-Practicum/feature/add_endp…

…oints_for_tech_messages

Добавление эндпоинтов для технических сообщений от бота

src/api/endpoints/__init__.py

@@ -6,6 +6,7 @@
 from .health_check import health_check_router
 from .notification import notification_router_by_admin, notification_router_by_token
 from .tasks import task_read_router, task_response_router, task_write_router, tasks_router
+from .tech_messages import tech_message_router
 from .telegram_webhook import telegram_webhook_router
 from .users import user_router
 
@@ -27,4 +28,5 @@
     "admin_user_router",
     "feedback_router",
     "user_router",
+    "tech_message_router",
 )

src/api/endpoints/admin/__init__.py

@@ -2,6 +2,7 @@
 
 from src.api.endpoints.analytics import analytic_router
 from src.api.endpoints.notification import notification_router_by_admin
+from src.api.endpoints.tech_messages import tech_message_router
 from src.api.endpoints.users import user_router
 from src.api.fastapi_admin_users import auth_backend, auth_cookie_backend, fastapi_admin_users
 from src.api.permissions import is_active_user
@@ -18,7 +19,7 @@
 admin_router.include_router(notification_router_by_admin, prefix="/messages", tags=["Messages"])
 admin_router.include_router(user_router, prefix="/users", tags=["User"])
 admin_router.include_router(admin_user_list_router, prefix="/admins", tags=["Admins"])
-
+admin_router.include_router(tech_message_router, prefix="/tech_messages", tags=["Tech messages"])
 
 admin_auth_router = APIRouter()
 admin_auth_router.include_router(fastapi_admin_users.get_auth_router(auth_backend))

src/api/endpoints/tech_messages.py

@@ -0,0 +1,75 @@
+from dependency_injector.wiring import Provide, inject
+from fastapi import APIRouter, Depends, Query, Request, status
+
+from src.api.pagination import TechMessagePaginator
+from src.api.permissions import is_active_superuser
+from src.api.schemas import TechMessagePaginateResponse, TechMessageRequest, TechMessageResponce
+from src.core.depends import Container
+from src.core.services import TechMessageService
+
+tech_message_router = APIRouter()
+
+
+@tech_message_router.get(
+    "",
+    response_model=TechMessagePaginateResponse,
+    response_model_exclude_none=True,
+    description="Получает список технических сообщений.",
+)
+@inject
+async def get_all_tech_messages(
+    request: Request,
+    was_read: bool | None = None,
+    page: int = Query(default=1, ge=1),
+    limit: int = Query(default=20, ge=1),
+    tech_message_service: TechMessageService = Depends(Provide[Container.core_services_container.tech_message]),
+    tech_message_paginate: TechMessagePaginator = Depends(
+        Provide[Container.api_paginate_container.tech_message_paginate]
+    ),
+) -> TechMessagePaginateResponse:
+    filter_by = {"was_read": was_read}
+    tech_messages = await tech_message_service.get_filtered_tech_messages_by_page(filter_by, page, limit)
+    return await tech_message_paginate.paginate(tech_messages, page, limit, request.url.path, filter_by)
+
+
+@tech_message_router.get(
+    "/{message_id}",
+    response_model=TechMessageResponce,
+    response_model_exclude_none=True,
+    description="Получает техническое сообщение.",
+)
+@inject
+async def get_tech_message(
+    message_id: int,
+    tech_message_service: TechMessageService = Depends(Provide[Container.core_services_container.tech_message]),
+) -> TechMessageResponce:
+    return await tech_message_service.get(message_id)
+
+
+@tech_message_router.patch(
+    "/{message_id}",
+    response_model=TechMessageResponce,
+    response_model_exclude_none=True,
+    description="Обновляет данные технического сообщения.",
+)
+@inject
+async def patch_tech_message(
+    message_id: int,
+    tech_message_request: TechMessageRequest,
+    tech_message_service: TechMessageService = Depends(Provide[Container.core_services_container.tech_message]),
+) -> TechMessageResponce:
+    return await tech_message_service.partial_update(message_id, tech_message_request.model_dump())
+
+
+@tech_message_router.delete(
+    "/{message_id}",
+    status_code=status.HTTP_204_NO_CONTENT,
+    description="Архивирует техническое сообщение.",
+    dependencies=[Depends(is_active_superuser)],
+)
+@inject
+async def delete_tech_message(
+    message_id: int,
+    tech_message_service: TechMessageService = Depends(Provide[Container.core_services_container.tech_message]),
+) -> None:
+    return await tech_message_service.archive(message_id)

src/api/pagination.py

@@ -1,8 +1,8 @@
 import math
 from typing import Any, Generic, TypeVar
 
-from src.core.db.models import AdminUser, User
-from src.core.db.repository import AbstractRepository, AdminUserRepository, UserRepository
+from src.core.db.models import AdminUser, TechMessage, User
+from src.core.db.repository import AbstractRepository, AdminUserRepository, TechMessageRepository, UserRepository
 from src.core.db.repository.base import FilterableRepository
 
 DatabaseModel = TypeVar("DatabaseModel")
@@ -67,3 +67,10 @@ class UserPaginator(FilterablePaginator[User]):
 
     def __init__(self, user_repository: UserRepository) -> None:
         super().__init__(user_repository)
+
+
+class TechMessagePaginator(FilterablePaginator[TechMessage]):
+    """Класс для пагинации и фильтрации данных из модели TechMessage."""
+
+    def __init__(self, repository: TechMessageRepository) -> None:
+        super().__init__(repository)

src/api/schemas/__init__.py

@@ -21,6 +21,7 @@
     TelegramNotificationUsersRequest,
 )
 from .tasks import TaskRequest, TaskResponse, TasksRequest, UserResponseToTaskRequest
+from .tech_messages import TechMessagePaginateResponse, TechMessageRequest, TechMessageResponce
 from .token_schemas import TokenCheckResponse
 from .users import UserResponse, UsersPaginatedResponse
 
@@ -59,4 +60,7 @@
     "UserResponse",
     "UserResponseToTaskRequest",
     "UsersPaginatedResponse",
+    "TechMessageResponce",
+    "TechMessageRequest",
+    "TechMessagePaginateResponse",
 )

src/api/schemas/tech_messages.py

@@ -0,0 +1,24 @@
+from datetime import datetime
+
+from src.api.schemas.base import PaginateBase, RequestBase, ResponseBase
+
+
+class TechMessageResponce(ResponseBase):
+    """Класс схемы ответа."""
+
+    id: int
+    text: str
+    was_read: bool
+    created_at: datetime
+
+
+class TechMessageRequest(RequestBase):
+    """Класс схемы запроса."""
+
+    was_read: bool
+
+
+class TechMessagePaginateResponse(PaginateBase):
+    """Класс схемы постраничного ответа."""
+
+    result: list[TechMessageResponce] | None

src/core/db/repository/base.py

@@ -146,6 +146,13 @@ async def get(self, id: int, *, is_archived: bool | None = False) -> DatabaseMod
             raise NotFoundException(object_name=self._model.__name__, object_id=id)
         return db_obj
 
+    async def archive(self, id: int) -> DatabaseModel:
+        """Архивирует объект модели"""
+        db_obj = await self.get(id)
+        db_obj.is_archived = True
+        db_obj = await self.update(id, db_obj)
+        return db_obj
+
 
 class ContentRepository(ArchivableRepository):
     """Абстрактный класс, для контента."""

src/core/db/repository/tech_message.py

@@ -1,3 +1,7 @@
+from collections.abc import Sequence
+from typing import Any
+
+from sqlalchemy import Select, desc, false, func, select, true
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from src.core.db.models import TechMessage
@@ -9,3 +13,45 @@ class TechMessageRepository(ArchivableRepository):
 
     def __init__(self, session: AsyncSession) -> None:
         super().__init__(session, TechMessage)
+
+    async def partial_update(self, id: int, data: dict) -> TechMessage:
+        """Обновляет данные/часть данных технического сообщения."""
+        tech_message = await self.get(id)
+
+        for attr, value in data.items():
+            setattr(tech_message, attr, value)
+
+        return await self.update(id, tech_message)
+
+    def _add_filter_by_was_read(self, statement: Select, was_read: bool | None) -> Select:
+        """Добавляет к оператору SELECT проверку статуса технического сообщения (was_read)."""
+        if was_read is True:
+            return statement.where(TechMessage.was_read == true())
+        elif was_read is False:
+            return statement.where(TechMessage.was_read == false())
+        return statement
+
+    async def count_by_filter(self, filter_by: dict) -> int:
+        """Возвращает количество не архивных данных, удовлетворяющих фильтру."""
+        statement = self._add_filter_by_was_read(
+            select(func.count()).select_from(TechMessage),
+            filter_by.get("was_read"),
+        )
+        return await self._session.scalar(statement.where(TechMessage.is_archived == false()))
+
+    async def get_filtered_tech_messages_by_page(
+        self, filter_by: dict[str:Any], page: int, limit: int, column_name: str = "created_at"
+    ) -> Sequence[TechMessage]:
+        """
+        Получает отфильтрованные не архивные данные, ограниченные параметрами page и limit
+        и отсортированные по полю column_name в порядке убывания.
+        """
+        offset = (page - 1) * limit
+        statement = self._add_filter_by_was_read(
+            select(TechMessage),
+            filter_by.get("was_read"),
+        )
+        objects = await self._session.scalars(
+            statement.where(TechMessage.is_archived == false()).limit(limit).offset(offset).order_by(desc(column_name))
+        )
+        return objects.all()

src/core/depends/pagination.py

@@ -1,6 +1,6 @@
 from dependency_injector import containers, providers
 
-from src.api.pagination import AdminUserPaginator, UserPaginator
+from src.api.pagination import AdminUserPaginator, TechMessagePaginator, UserPaginator
 
 
 class PaginateContainer(containers.DeclarativeContainer):
@@ -17,3 +17,8 @@ class PaginateContainer(containers.DeclarativeContainer):
         AdminUserPaginator,
         repository=repositories.admin_repository,
     )
+
+    tech_message_paginate = providers.Factory(
+        TechMessagePaginator,
+        repository=repositories.tech_message_repository,
+    )

src/core/services/tech_message.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 from src.core.db.models import TechMessage
 from src.core.db.repository import TechMessageRepository
 
@@ -10,3 +12,17 @@ def __init__(self, repository: TechMessageRepository):
 
     async def create(self, message: str) -> TechMessage:
         return await self._repository.create(TechMessage(text=message))
+
+    async def get_filtered_tech_messages_by_page(
+        self, filter_by: dict[str:Any], page: int, limit: int
+    ) -> list[TechMessage]:
+        return await self._repository.get_filtered_tech_messages_by_page(filter_by, page, limit)
+
+    async def get(self, id: int) -> TechMessage:
+        return await self._repository.get(id)
+
+    async def partial_update(self, id: int, data: dict) -> TechMessage:
+        return await self._repository.partial_update(id, data)
+
+    async def archive(self, id: int) -> None:
+        await self._repository.archive(id)