Feat/bot catalog (#408)

* feat: bot catalog method

* bump: version

pybotx/__init__.py

@@ -62,6 +62,7 @@
     OutgoingAttachment,
 )
 from pybotx.models.bot_account import BotAccount, BotAccountWithSecret
+from pybotx.models.bot_catalog import BotsListItem
 from pybotx.models.bot_sender import BotSender
 from pybotx.models.chats import Chat, ChatInfo, ChatInfoMember, ChatListItem
 from pybotx.models.enums import (
@@ -199,6 +200,7 @@
     "SmartAppEvent",
     "SmartAppEvent",
     "StatusRecipient",
+    "BotsListItem",
     "StealthModeDisabledError",
     "Sticker",
     "StickerPack",

pybotx/bot/bot.py

@@ -1,5 +1,6 @@
 from asyncio import Task
 from contextlib import asynccontextmanager
+from datetime import datetime
 from types import SimpleNamespace
 from typing import (
     Any,
@@ -29,6 +30,10 @@
 from pybotx.bot.handler import Middleware
 from pybotx.bot.handler_collector import HandlerCollector
 from pybotx.bot.middlewares.exception_middleware import ExceptionHandlersDict
+from pybotx.client.bots_api.bot_catalog import (
+    BotsListMethod,
+    BotXAPIBotsListRequestPayload,
+)
 from pybotx.client.chats_api.add_admin import (
     AddAdminMethod,
     BotXAPIAddAdminRequestPayload,
@@ -200,6 +205,7 @@
 from pybotx.models.async_files import File
 from pybotx.models.attachments import IncomingFileAttachment, OutgoingAttachment
 from pybotx.models.bot_account import BotAccount, BotAccountWithSecret
+from pybotx.models.bot_catalog import BotsListItem
 from pybotx.models.chats import ChatInfo, ChatListItem
 from pybotx.models.commands import BotAPICommand, BotCommand
 from pybotx.models.enums import ChatTypes
@@ -376,6 +382,31 @@ async def get_token(
 
         return await get_token(bot_id, self._httpx_client, self._bot_accounts_storage)
 
+    async def get_bots_list(
+        self,
+        *,
+        bot_id: UUID,
+        since: Missing[datetime] = Undefined,
+    ) -> Tuple[List[BotsListItem], datetime]:
+        """Get list of Bots on the current CTS.
+
+        :param bot_id: Bot which should perform the request.
+        :param since: Only return bots changed after this date.
+
+        :return: List of Bots, generated timestamp.
+        """
+
+        method = BotsListMethod(
+            bot_id,
+            self._httpx_client,
+            self._bot_accounts_storage,
+        )
+        payload = BotXAPIBotsListRequestPayload.from_domain(since=since)
+
+        botx_api_bots_list = await method.execute(payload)
+
+        return botx_api_bots_list.to_domain()
+
     # - Notifications API -
     async def answer_message(
         self,

pybotx/client/bots_api/bot_catalog.py

@@ -0,0 +1,69 @@
+from datetime import datetime
+from typing import List, Literal, Optional, Tuple
+from uuid import UUID
+
+from pybotx.client.authorized_botx_method import AuthorizedBotXMethod
+from pybotx.missing import Missing, Undefined
+from pybotx.models.api_base import UnverifiedPayloadBaseModel, VerifiedPayloadBaseModel
+from pybotx.models.bot_catalog import BotsListItem
+
+
+class BotXAPIBotsListRequestPayload(UnverifiedPayloadBaseModel):
+    since: Missing[datetime] = Undefined
+
+    @classmethod
+    def from_domain(
+        cls,
+        since: Missing[datetime] = Undefined,
+    ) -> "BotXAPIBotsListRequestPayload":
+        return cls(since=since)
+
+
+class BotXAPIBotItem(VerifiedPayloadBaseModel):
+    user_huid: UUID
+    name: str
+    description: str
+    avatar: Optional[str] = None
+    enabled: bool
+
+
+class BotXAPIBotsListResult(VerifiedPayloadBaseModel):
+    generated_at: datetime
+    bots: List[BotXAPIBotItem]
+
+
+class BotXAPIBotsListResponsePayload(VerifiedPayloadBaseModel):
+    result: BotXAPIBotsListResult
+    status: Literal["ok"]
+
+    def to_domain(self) -> Tuple[List[BotsListItem], datetime]:
+        bots_list = [
+            BotsListItem(
+                id=bot.user_huid,
+                name=bot.name,
+                description=bot.description,
+                avatar=bot.avatar,
+                enabled=bot.enabled,
+            )
+            for bot in self.result.bots
+        ]
+        return bots_list, self.result.generated_at
+
+
+class BotsListMethod(AuthorizedBotXMethod):
+    async def execute(
+        self,
+        payload: BotXAPIBotsListRequestPayload,
+    ) -> BotXAPIBotsListResponsePayload:
+        path = "/api/v1/botx/bots/catalog"
+
+        response = await self._botx_method_call(
+            "GET",
+            self._build_url(path),
+            params=payload.jsonable_dict(),
+        )
+
+        return self._verify_and_extract_api_model(
+            BotXAPIBotsListResponsePayload,
+            response,
+        )

pybotx/models/bot_catalog.py

@@ -0,0 +1,22 @@
+from dataclasses import dataclass
+from typing import Optional
+from uuid import UUID
+
+
+@dataclass
+class BotsListItem:
+    """Bot from list of bots.
+
+    Attributes:
+        id: Bot user huid.
+        name: Bot name.
+        description: Bot description.
+        avatar: Bot avatar url.
+        enabled: Is the SmartApp enabled or not.
+    """
+
+    id: UUID
+    name: str
+    description: str
+    avatar: Optional[str]
+    enabled: bool

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "pybotx"
-version = "0.58.0"
+version = "0.59.0"
 description = "A python library for interacting with eXpress BotX API"
 authors = [
     "Sidnev Nikolay <[email protected]>",

tests/client/bots_api/test_bots_list.py

@@ -0,0 +1,84 @@
+from datetime import datetime
+from http import HTTPStatus
+from uuid import UUID
+
+import httpx
+import pytest
+from respx import MockRouter
+
+from pybotx import Bot, BotAccountWithSecret, HandlerCollector, lifespan_wrapper
+from pybotx.models.bot_catalog import BotsListItem
+
+pytestmark = [
+    pytest.mark.asyncio,
+    pytest.mark.mock_authorization,
+    pytest.mark.usefixtures("respx_mock"),
+]
+
+
+async def test__smartapps_list__succeed(
+    respx_mock: MockRouter,
+    host: str,
+    bot_id: UUID,
+    bot_account: BotAccountWithSecret,
+) -> None:
+    # - Arrange -
+    endpoint = respx_mock.get(
+        f"https://{host}/api/v1/botx/bots/catalog",
+        headers={"Authorization": "Bearer token"},
+    ).mock(
+        return_value=httpx.Response(
+            HTTPStatus.OK,
+            json={
+                "result": {
+                    "generated_at": datetime(2023, 1, 1).isoformat(),
+                    "bots": [
+                        {
+                            "user_huid": "6fafda2c-6505-57a5-a088-25ea5d1d0364",
+                            "name": "First bot",
+                            "description": "My bot",
+                            "avatar": None,
+                            "enabled": True,
+                        },
+                        {
+                            "user_huid": "66d74e0a-b3c8-4c28-a03f-baf2d1d3f4c7",
+                            "name": "Second bot",
+                            "description": "Your bot",
+                            "avatar": "https://cts.example.com/uploads/profile_avatar/bar",
+                            "enabled": True,
+                        },
+                    ],
+                },
+                "status": "ok",
+            },
+        ),
+    )
+
+    built_bot = Bot(collectors=[HandlerCollector()], bot_accounts=[bot_account])
+
+    # - Act -
+    async with lifespan_wrapper(built_bot) as bot:
+        bots_list, timestamp = await bot.get_bots_list(
+            bot_id=bot_id,
+            since=datetime(2022, 1, 1),
+        )
+
+    # - Assert -
+    assert endpoint.called
+    assert timestamp == datetime(2023, 1, 1)
+    assert bots_list == [
+        BotsListItem(
+            id=UUID("6fafda2c-6505-57a5-a088-25ea5d1d0364"),
+            name="First bot",
+            description="My bot",
+            avatar=None,
+            enabled=True,
+        ),
+        BotsListItem(
+            id=UUID("66d74e0a-b3c8-4c28-a03f-baf2d1d3f4c7"),
+            name="Second bot",
+            description="Your bot",
+            avatar="https://cts.example.com/uploads/profile_avatar/bar",
+            enabled=True,
+        ),
+    ]