Init project

birthday_pool_bot/telegram_bot/webhook/__init__.py

@@ -0,0 +1,10 @@
+from .service import TelegramBotWebhookService
+from .settings import TelegramBotWebhookSettings
+
+
+__all__ = (
+    # service
+    "TelegramBotWebhookService",
+    # settings
+    "TelegramBotWebhookSettings",
+)

birthday_pool_bot/telegram_bot/webhook/service.py

@@ -0,0 +1,49 @@
+from typing import Any
+
+import aiogram
+import fastapi
+import uvicorn
+
+from birthday_pool_bot.telegram_bot.base import BaseTelegramBotService
+
+
+class UvicornServer(uvicorn.Server):
+    def install_signal_handlers(self):
+        pass
+
+
+class TelegramBotWebhookService(BaseTelegramBotService):
+    async def listen_events(self):
+        await self._bot.set_webhook(
+            url=self._settings.url,
+            secret_token=self._settings.secret_token,
+        )
+
+        await self.get_server().serve()
+
+    def get_server(self) -> uvicorn.Server:
+        app = fastapi.FastAPI(
+            root_url=self._settings.root_url,
+            root_path=self._settings.root_path,
+        )
+
+        app.post(self._webhook_path)(self.handler)
+
+        config = uvicorn.Config(app=app, host="0.0.0.0", port=self._port)
+        return UvicornServer(config)
+
+    async def handler(
+            self,
+            update: dict[str, Any],
+            x_telegram_bot_api_secret_token: str | None = fastapi.Header(None),
+    ):
+        if x_telegram_bot_api_secret_token != self._secret_token:
+            raise fastapi.HTTPException(
+                status_code=fastapi.status.HTTP_403_FORBIDDEN,
+                detail="Forbidden.",
+            )
+
+        await self._dispatcher.feed_webhook_update(
+            bot=self._bot,
+            update=aiogram.types.Update(**update),
+        )

birthday_pool_bot/telegram_bot/webhook/settings.py

@@ -0,0 +1,15 @@
+from typing import Literal
+
+from pydantic import AnyHttpUrl, conint, constr
+
+from birthday_pool_bot.telegram_bot.enums import TelegramBotMethodEnum
+from birthday_pool_bot.telegram_bot.settings import TelegramBotSettings
+
+
+class TelegramBotWebhookSettings(TelegramBotSettings):
+    method: Literal[TelegramBotMethodEnum.WEBHOOK.value] = TelegramBotMethodEnum.WEBHOOK
+
+    root_url: AnyHttpUrl
+    root_path: str = "/"
+    port: conint(ge=1, le=65535) = 8000
+    secret_access_key: constr(pattern=r"^\w{32}$") = "webhooksuperpupersecretaccesskey"