Merge pull request #168 from Studio-Yandex-Practicum/feature/add_webhook

feature/add_webhook

src/api/endpoints/__init__.py

@@ -3,11 +3,13 @@
 from .external_site_user import site_user_router
 from .notification import notification_router
 from .tasks import task_router
+from .telegram_webhook import telegram_webhook_router
 
 __all__ = (
     "analytic_router",
     "category_router",
     "task_router",
+    "telegram_webhook_router",
     "form_router",
     "notification_router",
     "site_user_router",

src/api/endpoints/telegram_webhook.py

@@ -0,0 +1,23 @@
+from fastapi import APIRouter, Request
+from telegram import Update
+
+from src.core.exceptions.exceptions import UnauthorizedError, WebhookOnError
+from src.settings import settings
+
+telegram_webhook_router = APIRouter()
+
+
+@telegram_webhook_router.post(
+    "/webhook",
+    description="Получить обновления telegram.",
+)
+async def get_telegram_bot_updates(request: Request) -> None:
+    """Получение обновлений telegram в режиме работы бота webhook."""
+    if not settings.BOT_WEBHOOK_MODE:
+        raise WebhookOnError
+    secret_token = request.headers.get("x-telegram-bot-api-secret-token")
+    if secret_token != settings.SECRET_KEY:
+        raise UnauthorizedError
+    bot_instance = request.app.state.bot_instance
+    request_json_data = await request.json()
+    await bot_instance.update_queue.put(Update.de_json(data=request_json_data, bot=bot_instance.bot))

src/api/router.py

@@ -1,11 +1,19 @@
 from fastapi import APIRouter
 
-from src.api.endpoints import analytic_router, category_router, notification_router, site_user_router, task_router
+from src.api.endpoints import (
+    analytic_router,
+    category_router,
+    notification_router,
+    site_user_router,
+    task_router,
+    telegram_webhook_router,
+)
 from src.settings import settings
 
 api_router = APIRouter(prefix=settings.ROOT_PATH)
-api_router.include_router(category_router, prefix="/categories", tags=["Categories"])
 api_router.include_router(analytic_router, prefix="/analytics", tags=["Analytic"])
-api_router.include_router(task_router, prefix="/tasks", tags=["Tasks"])
+api_router.include_router(category_router, prefix="/categories", tags=["Categories"])
 api_router.include_router(notification_router, prefix="/messages", tags=["Messages"])
 api_router.include_router(site_user_router, prefix="/external_user_registration", tags=["ExternalSiteUser"])
+api_router.include_router(task_router, prefix="/tasks", tags=["Tasks"])
+api_router.include_router(telegram_webhook_router, prefix="/telegram", tags=["Telegram"])

src/core/exceptions/exceptions.py

@@ -33,3 +33,13 @@ class EmailSendError(ApplicationException):
 
     def __init__(self, recipients: EmailStr | list[EmailStr], exc: Exception):
         self.detail = f"Возникла ошибка {exc} при отправке email на адрес {recipients}."
+
+
+class UnauthorizedError(ApplicationException):
+    status_code: HTTPStatus = HTTPStatus.UNAUTHORIZED
+    detail = "У Вас нет прав для просмотра запрошенной страницы."
+
+
+class WebhookOnError(ApplicationException):
+    status_code: HTTPStatus = HTTPStatus.NO_CONTENT
+    detail = "Telegram Webhook выключен."