Feature/voex call api

pybotx/bot/bot.py

@@ -224,6 +224,14 @@
     BotXAPIUsersAsCSVRequestPayload,
     UsersAsCSVMethod,
 )
+from pybotx.client.voex_api.get_call import (
+    BotXAPIGetCallRequestPayload,
+    GetCallMethod,
+)
+from pybotx.client.voex_api.get_conference import (
+    BotXAPIGetConferenceRequestPayload,
+    GetConferenceMethod,
+)
 from pybotx.constants import BOTX_DEFAULT_TIMEOUT, STICKER_PACKS_PER_PAGE
 from pybotx.converters import optional_sequence_to_list
 from pybotx.image_validators import (
@@ -236,12 +244,14 @@
 from pybotx.models.attachments import IncomingFileAttachment, OutgoingAttachment
 from pybotx.models.bot_account import BotAccountWithSecret
 from pybotx.models.bot_catalog import BotsListItem
+from pybotx.models.call import Call
 from pybotx.models.chats import ChatInfo, ChatLink, ChatListItem
 from pybotx.models.commands import (
     BotAPISystemEvent,
     BotAPIIncomingMessage,
     BotCommand,
 )
+from pybotx.models.conference import Conference
 from pybotx.models.enums import BotAPICommandTypes, ChatLinkTypes, ChatTypes
 from pybotx.models.message.edit_message import EditMessage
 from pybotx.models.message.markup import BubbleMarkup, KeyboardMarkup
@@ -1395,6 +1405,56 @@ async def pin_message(
 
         await method.execute(payload)
 
+    async def get_call(
+        self,
+        *,
+        bot_id: UUID,
+        call_id: UUID,
+    ) -> Call:
+        """Get call.
+
+        :param bot_id: Bot which should perform the request.
+        :param call_id: Call id.
+
+        :return: Call.
+        """
+        method = GetCallMethod(
+            bot_id,
+            self._httpx_client,
+            self._bot_accounts_storage,
+        )
+        payload = BotXAPIGetCallRequestPayload.from_domain(
+            call_id=call_id,
+        )
+        botx_call = await method.execute(payload)
+
+        return botx_call.to_domain()
+
+    async def get_conference(
+        self,
+        *,
+        bot_id: UUID,
+        call_id: UUID,
+    ) -> Conference:
+        """Get Conference.
+
+        :param bot_id: Bot which should perform the request.
+        :param call_id: Call id.
+
+        :return: Conference.
+        """
+        method = GetConferenceMethod(
+            bot_id,
+            self._httpx_client,
+            self._bot_accounts_storage,
+        )
+        payload = BotXAPIGetConferenceRequestPayload.from_domain(
+            call_id=call_id,
+        )
+        botx_conference = await method.execute(payload)
+
+        return botx_conference.to_domain()
+
     async def unpin_message(
         self,
         *,

pybotx/client/voex_api/exceptions.py

@@ -0,0 +1,9 @@
+from pybotx.client.exceptions.base import BaseClientError
+
+
+class ConferenceNotFoundError(BaseClientError):
+    """Conference with specified call_id not found."""
+
+
+class CallNotFoundError(BaseClientError):
+    """Call with specified call_id not found."""

pybotx/client/voex_api/get_call.py

@@ -0,0 +1,59 @@
+from typing import Literal
+from uuid import UUID
+
+from pybotx.client.authorized_botx_method import AuthorizedBotXMethod
+from pybotx.client.botx_method import response_exception_thrower
+from pybotx.client.voex_api.exceptions import CallNotFoundError
+from pybotx.models.api_base import UnverifiedPayloadBaseModel, VerifiedPayloadBaseModel
+from pybotx.models.call import Call
+
+
+class BotXAPIGetCallRequestPayload(UnverifiedPayloadBaseModel):
+    call_id: UUID
+
+    @classmethod
+    def from_domain(
+        cls,
+        call_id: UUID,
+    ) -> "BotXAPIGetCallRequestPayload":
+        return cls(call_id=call_id)
+
+
+class BotXAPIGetCallResult(VerifiedPayloadBaseModel):
+    id: UUID
+    members: list[UUID]
+
+
+class BotXAPIGetCallResponsePayload(VerifiedPayloadBaseModel):
+    status: Literal["ok"]
+    result: BotXAPIGetCallResult
+
+    def to_domain(self) -> Call:
+        return Call(
+            id=self.result.id,
+            members=self.result.members,
+        )
+
+
+class GetCallMethod(AuthorizedBotXMethod):
+    status_handlers = {
+        **AuthorizedBotXMethod.status_handlers,
+        404: response_exception_thrower(CallNotFoundError),
+    }
+
+    async def execute(
+        self,
+        payload: BotXAPIGetCallRequestPayload,
+    ) -> BotXAPIGetCallResponsePayload:
+        jsonable_dict = payload.jsonable_dict()
+        path = f"/api/v3/botx/voex/calls/{jsonable_dict['call_id']}"
+
+        response = await self._botx_method_call(
+            "GET",
+            self._build_url(path),
+        )
+
+        return self._verify_and_extract_api_model(
+            BotXAPIGetCallResponsePayload,
+            response,
+        )

pybotx/client/voex_api/get_conference.py

@@ -0,0 +1,63 @@
+from typing import Literal
+from uuid import UUID
+
+from pybotx.client.authorized_botx_method import AuthorizedBotXMethod
+from pybotx.client.botx_method import response_exception_thrower
+from pybotx.client.voex_api.exceptions import ConferenceNotFoundError
+from pybotx.models.api_base import UnverifiedPayloadBaseModel, VerifiedPayloadBaseModel
+from pybotx.models.conference import Conference
+
+
+class BotXAPIGetConferenceRequestPayload(UnverifiedPayloadBaseModel):
+    call_id: UUID
+
+    @classmethod
+    def from_domain(
+        cls,
+        call_id: UUID,
+    ) -> "BotXAPIGetConferenceRequestPayload":
+        return cls(call_id=call_id)
+
+
+class BotXAPIGetConferenceResult(VerifiedPayloadBaseModel):
+    id: UUID
+    name: str
+    link: str
+    members: list[UUID]
+
+
+class BotXAPIGetConferenceResponsePayload(VerifiedPayloadBaseModel):
+    status: Literal["ok"]
+    result: BotXAPIGetConferenceResult
+
+    def to_domain(self) -> Conference:
+        return Conference(
+            id=self.result.id,
+            name=self.result.name,
+            link=self.result.link,
+            members=self.result.members,
+        )
+
+
+class GetConferenceMethod(AuthorizedBotXMethod):
+    status_handlers = {
+        **AuthorizedBotXMethod.status_handlers,
+        404: response_exception_thrower(ConferenceNotFoundError),
+    }
+
+    async def execute(
+        self,
+        payload: BotXAPIGetConferenceRequestPayload,
+    ) -> BotXAPIGetConferenceResponsePayload:
+        jsonable_dict = payload.jsonable_dict()
+        path = f"/api/v3/botx/voex/conferences/{jsonable_dict['call_id']}"
+
+        response = await self._botx_method_call(
+            "GET",
+            self._build_url(path),
+        )
+
+        return self._verify_and_extract_api_model(
+            BotXAPIGetConferenceResponsePayload,
+            response,
+        )

pybotx/models/call.py

@@ -0,0 +1,8 @@
+from dataclasses import dataclass
+from uuid import UUID
+
+
+@dataclass(slots=True)
+class Call:
+    id: UUID
+    members: list[UUID]

pybotx/models/conference.py

@@ -0,0 +1,10 @@
+from dataclasses import dataclass
+from uuid import UUID
+
+
+@dataclass(slots=True)
+class Conference:
+    id: UUID
+    name: str
+    link: str
+    members: list[UUID]

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "pybotx"
-version = "0.76.3"
+version = "0.76.4"
 description = "A python library for interacting with eXpress BotX API"
 authors = [
     "Sidnev Nikolay <nsidnev@ccsteam.ru>",

tests/client/voex_api/test_call.py

@@ -0,0 +1,113 @@
+from http import HTTPStatus
+from uuid import UUID
+
+import httpx
+import pytest
+from respx.router import MockRouter
+
+from pybotx import (
+    Bot,
+    BotAccountWithSecret,
+    HandlerCollector,
+    lifespan_wrapper,
+)
+from pybotx.client.voex_api.exceptions import CallNotFoundError
+from pybotx.models.call import Call
+
+pytestmark = [
+    pytest.mark.asyncio,
+    pytest.mark.mock_authorization,
+    pytest.mark.usefixtures("respx_mock"),
+]
+
+
+async def test__get_call__succeed(
+    respx_mock: MockRouter,
+    host: str,
+    bot_id: UUID,
+    bot_account: BotAccountWithSecret,
+) -> None:
+    # - Arrange -
+    call_id = "a465f0f3-1354-491c-8f11-f400164295cb"
+    member1 = "dcfa5a7c-7cc4-4c89-b6c0-80325604f9f4"
+    member2 = "6fa5f1e9-1453-0ad7-2d6d-b791467e382a"
+
+    endpoint = respx_mock.get(
+        f"https://{host}/api/v3/botx/voex/calls/{call_id}",
+        headers={"Authorization": "Bearer token"},
+    ).mock(
+        return_value=httpx.Response(
+            HTTPStatus.OK,
+            json={
+                "status": "ok",
+                "result": {
+                    "id": call_id,
+                    "members": [
+                        member1,
+                        member2,
+                    ],
+                },
+            },
+        ),
+    )
+
+    built_bot = Bot(collectors=[HandlerCollector()], bot_accounts=[bot_account])
+
+    # - Act -
+    async with lifespan_wrapper(built_bot) as bot:
+        call = await bot.get_call(
+            bot_id=bot_id,
+            call_id=UUID(call_id),
+        )
+
+    # - Assert -
+    assert call == Call(
+        id=UUID(call_id),
+        members=[
+            UUID(member1),
+            UUID(member2),
+        ],
+    )
+
+    assert endpoint.called
+
+
+async def test__get_call__call_not_found(
+    respx_mock: MockRouter,
+    host: str,
+    bot_id: UUID,
+    bot_account: BotAccountWithSecret,
+) -> None:
+    # - Arrange -
+    call_id = "a465f0f3-1354-491c-8f11-f400164295cb"
+
+    endpoint = respx_mock.get(
+        f"https://{host}/api/v3/botx/voex/calls/{call_id}",
+        headers={"Authorization": "Bearer token"},
+    ).mock(
+        return_value=httpx.Response(
+            HTTPStatus.NOT_FOUND,
+            json={
+                "error_data": {
+                    "call_id": call_id,
+                },
+                "errors": ["Call with specified call_id not found."],
+                "reason": "not_found",
+                "status": "error",
+            },
+        ),
+    )
+
+    built_bot = Bot(collectors=[HandlerCollector()], bot_accounts=[bot_account])
+
+    # - Act -
+    async with lifespan_wrapper(built_bot) as bot:
+        with pytest.raises(CallNotFoundError) as exc:
+            await bot.get_call(
+                bot_id=bot_id,
+                call_id=UUID(call_id),
+            )
+
+    # - Assert -
+    assert "not_found" in str(exc.value)
+    assert endpoint.called