feat: /api/v4/botx/smartapps/notification support

pybotx/bot/bot.py

@@ -118,6 +118,10 @@
     BotXAPIRefreshAccessTokenRequestPayload,
     RefreshAccessTokenMethod,
 )
+from pybotx.client.smartapps_api.smartapp_custom_notification import (
+    BotXAPISmartAppCustomNotificationRequestPayload,
+    SmartAppCustomNotificationMethod,
+)
 from pybotx.client.smartapps_api.smartapp_event import (
     BotXAPISmartAppEventRequestPayload,
     SmartAppEventMethod,
@@ -1475,6 +1479,53 @@ async def upload_static_file(
 
         return botx_api_static_file.to_domain()
 
+    async def send_smartapp_custom_notification(
+        self,
+        *,
+        bot_id: UUID,
+        group_chat_id: UUID,
+        title: str,
+        body: str,
+        meta: Missing[Dict[str, Any]] = Undefined,
+        wait_callback: bool = True,
+        callback_timeout: Optional[float] = None,
+    ) -> UUID:
+        """Send SmartApp custom notification.
+
+        :param bot_id: Bot which should perform the request.
+        :param group_chat_id: Target chat id.
+        :param title: Notification title.
+        :param body: Notification body.
+        :param meta: Meta information.
+        :param wait_callback: Block method call until callback received.
+        :param callback_timeout: Callback timeout in seconds (or `None` for
+            endless waiting).
+
+        :return: Notification sync_id.
+        """
+
+        method = SmartAppCustomNotificationMethod(
+            bot_id,
+            self._httpx_client,
+            self._bot_accounts_storage,
+            self._callbacks_manager,
+        )
+        payload = BotXAPISmartAppCustomNotificationRequestPayload.from_domain(
+            group_chat_id=group_chat_id,
+            title=title,
+            body=body,
+            meta=meta,
+        )
+
+        botx_api_sync_id = await method.execute(
+            payload,
+            wait_callback,
+            callback_timeout,
+            self._default_callback_timeout,
+        )
+
+        return botx_api_sync_id.to_domain()
+
     # - Stickers API -
     async def create_sticker_pack(
         self,

pybotx/client/smartapps_api/smartapp_custom_notification.py

@@ -0,0 +1,81 @@
+from typing import Any, Dict, Literal, Optional
+from uuid import UUID
+
+from pybotx.client.authorized_botx_method import AuthorizedBotXMethod
+from pybotx.missing import Missing
+from pybotx.models.api_base import UnverifiedPayloadBaseModel, VerifiedPayloadBaseModel
+
+
+class BotXAPISmartAppCustomNotificationNestedPayload(UnverifiedPayloadBaseModel):
+    title: str
+    body: str
+    meta: Missing[Dict[str, Any]]
+
+
+class BotXAPISmartAppCustomNotificationRequestPayload(UnverifiedPayloadBaseModel):
+    group_chat_id: UUID
+    payload: BotXAPISmartAppCustomNotificationNestedPayload
+
+    @classmethod
+    def from_domain(
+        cls,
+        group_chat_id: UUID,
+        title: str,
+        body: str,
+        meta: Missing[Dict[str, Any]],
+    ) -> "BotXAPISmartAppCustomNotificationRequestPayload":
+        return cls(
+            group_chat_id=group_chat_id,
+            payload=BotXAPISmartAppCustomNotificationNestedPayload(
+                title=title,
+                body=body,
+                meta=meta,
+            ),
+        )
+
+
+class BotXAPISyncIdResult(VerifiedPayloadBaseModel):
+    sync_id: UUID
+
+
+class BotXAPISmartAppCustomNotificationResponsePayload(VerifiedPayloadBaseModel):
+    status: Literal["ok"]
+    result: BotXAPISyncIdResult
+
+    def to_domain(self) -> UUID:
+        return self.result.sync_id
+
+
+class SmartAppCustomNotificationMethod(AuthorizedBotXMethod):
+    error_callback_handlers = {
+        **AuthorizedBotXMethod.error_callback_handlers,
+    }
+
+    async def execute(
+        self,
+        payload: BotXAPISmartAppCustomNotificationRequestPayload,
+        wait_callback: bool,
+        callback_timeout: Optional[float],
+        default_callback_timeout: float,
+    ) -> BotXAPISmartAppCustomNotificationResponsePayload:
+        path = "/api/v4/botx/smartapps/notification"
+
+        response = await self._botx_method_call(
+            "POST",
+            self._build_url(path),
+            json=payload.jsonable_dict(),
+        )
+
+        api_model = self._verify_and_extract_api_model(
+            BotXAPISmartAppCustomNotificationResponsePayload,
+            response,
+        )
+
+        await self._process_callback(
+            api_model.result.sync_id,
+            wait_callback,
+            callback_timeout,
+            default_callback_timeout,
+        )
+
+        return api_model

pyproject.toml

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

setup.cfg

@@ -57,6 +57,7 @@ per-file-ignores =
     # https://github.com/wemake-services/wemake-python-styleguide/issues/2172
     pybotx/bot/handler_collector.py:WPS437,
     pybotx/client/notifications_api/internal_bot_notification.py:WPS202,
+    pybotx/client/smartapps_api/smartapp_custom_notification.py:WPS118,
     # Complex model converting
     pybotx/models/message/incoming_message.py:WPS232,
     # WPS reacts at using `}` in f-strings

tests/client/smartapps_api/test_smartapp_custom_notification.py

@@ -0,0 +1,71 @@
+import asyncio
+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
+
+pytestmark = [
+    pytest.mark.asyncio,
+    pytest.mark.mock_authorization,
+    pytest.mark.usefixtures("respx_mock"),
+]
+
+
+async def test__send_smartapp_custom_notification__succeed(
+    respx_mock: MockRouter,
+    host: str,
+    bot_id: UUID,
+    bot_account: BotAccountWithSecret,
+) -> None:
+    # - Arrange -
+    endpoint = respx_mock.post(
+        f"https://{host}/api/v4/botx/smartapps/notification",
+        headers={"Authorization": "Bearer token", "Content-Type": "application/json"},
+        json={
+            "group_chat_id": "705df263-6bfd-536a-9d51-13524afaab5c",
+            "payload": {
+                "title": "test",
+                "body": "test",
+                "meta": {"message": "ping"},
+            },
+        },
+    ).mock(
+        return_value=httpx.Response(
+            HTTPStatus.ACCEPTED,
+            json={
+                "status": "ok",
+                "result": {"sync_id": "21a9ec9e-f21f-4406-ac44-1a78d2ccf9e3"},
+            },
+        ),
+    )
+
+    built_bot = Bot(collectors=[HandlerCollector()], bot_accounts=[bot_account])
+
+    # - Act -
+    async with lifespan_wrapper(built_bot) as bot:
+        task = asyncio.create_task(
+            bot.send_smartapp_custom_notification(
+                bot_id=bot_id,
+                group_chat_id=UUID("705df263-6bfd-536a-9d51-13524afaab5c"),
+                title="test",
+                body="test",
+                meta={"message": "ping"},
+            ),
+        )
+        await asyncio.sleep(0)  # Return control to event loop
+
+        await bot.set_raw_botx_method_result(
+            {
+                "status": "ok",
+                "sync_id": "21a9ec9e-f21f-4406-ac44-1a78d2ccf9e3",
+                "result": {},
+            },
+        )
+
+    # - Assert -
+    assert await task == UUID("21a9ec9e-f21f-4406-ac44-1a78d2ccf9e3")
+    assert endpoint.called