feat(workers): implement and register taskiq middlewares for logging, metrics, and retries

Author: PyDevDeep

backend/workers/broker.py

@@ -0,0 +1,26 @@
+from typing import Any
+
+from taskiq_redis import ListQueueBroker, RedisAsyncResultBackend
+
+from backend.config.settings import settings
+from backend.workers.middlewares.logging import StructlogMiddleware
+from backend.workers.middlewares.metrics import PrometheusMetricsMiddleware
+from backend.workers.middlewares.retry import RetryTrackerMiddleware
+
+redis_url = f"redis://{settings.REDIS_HOST}:{settings.REDIS_PORT}/0"
+
+# Ініціалізуємо брокер з явно вказаною чергою (ізоляція від інших проєктів)
+broker = ListQueueBroker(redis_url, queue_name="seratonin_tasks")
+
+# Додаємо явну типізацію для усунення помилок Pylance
+result_backend: RedisAsyncResultBackend[Any] = RedisAsyncResultBackend(
+    redis_url, result_ex_time=3600
+)
+broker.with_result_backend(result_backend)
+
+# Підключення Middlewares у правильному порядку
+broker.add_middlewares(
+    StructlogMiddleware(),
+    RetryTrackerMiddleware(),
+    PrometheusMetricsMiddleware(),
+)

backend/workers/dependencies.py

@@ -0,0 +1,40 @@
+from typing import Annotated
+
+from taskiq import TaskiqDepends
+
+from backend.integrations.llm.router import LLMRouter
+from backend.services.content_generator import ContentGenerator
+from backend.services.fact_checker import FactChecker
+from backend.services.style_matcher import StyleMatcher
+
+
+def get_llm_router() -> LLMRouter:
+    """Ініціалізує та повертає роутер моделей."""
+    return LLMRouter()
+
+
+def get_style_matcher() -> StyleMatcher:
+    """Ініціалізує сервіс пошуку стилю."""
+    return StyleMatcher()
+
+
+def get_fact_checker(
+    llm_router: Annotated[LLMRouter, TaskiqDepends(get_llm_router)],
+) -> FactChecker:
+    """Ініціалізує сервіс перевірки фактів, ін'єктуючи роутер."""
+    return FactChecker(llm_router=llm_router)
+
+
+def get_content_generator(
+    llm_router: Annotated[LLMRouter, TaskiqDepends(get_llm_router)],
+) -> ContentGenerator:
+    """
+    Фабрика для генератора контенту.
+    Використовує спільний інстанс роутера для оптимізації ресурсів.
+    """
+    return ContentGenerator(llm_router=llm_router)
+
+
+# Заглушка для Milestone 7.4 (Publisher Service)
+# def get_publisher_service() -> PublisherService:
+#     return PublisherService()

backend/workers/middlewares/logging.py

@@ -0,0 +1,34 @@
+from typing import Any
+
+import structlog
+from taskiq import TaskiqMessage, TaskiqMiddleware, TaskiqResult
+
+logger = structlog.get_logger()
+
+
+class StructlogMiddleware(TaskiqMiddleware):
+    def pre_execute(self, message: TaskiqMessage) -> TaskiqMessage:
+        """Виконується безпосередньо перед запуском таску у воркері."""
+        logger.info(
+            "task_execution_started",
+            task_id=message.task_id,
+            task_name=message.task_name,
+        )
+        return message
+
+    def post_execute(self, message: TaskiqMessage, result: TaskiqResult[Any]) -> None:
+        """Виконується після завершення таску (успішного або з помилкою)."""
+        if result.is_err:
+            logger.error(
+                "task_execution_failed",
+                task_id=message.task_id,
+                task_name=message.task_name,
+                error=str(result.error),
+            )
+        else:
+            logger.info(
+                "task_execution_success",
+                task_id=message.task_id,
+                task_name=message.task_name,
+                execution_time_sec=round(result.execution_time, 3),
+            )

backend/workers/middlewares/metrics.py

@@ -0,0 +1,21 @@
+from typing import Any
+
+import structlog
+from taskiq import TaskiqMessage, TaskiqMiddleware, TaskiqResult
+
+logger = structlog.get_logger()
+
+
+class PrometheusMetricsMiddleware(TaskiqMiddleware):
+    def post_execute(self, message: TaskiqMessage, result: TaskiqResult[Any]) -> None:
+        """
+        TODO: Інтеграція з prometheus_client.
+        Тут будуть оновлюватись Counter (кількість задач) та Histogram (час виконання).
+        """
+        status = "error" if result.is_err else "success"
+        logger.debug(
+            "metrics_updated",
+            metric="task_execution",
+            task_name=message.task_name,
+            status=status,
+        )

backend/workers/middlewares/retry.py

@@ -0,0 +1,20 @@
+import structlog
+from taskiq import TaskiqMessage, TaskiqMiddleware
+
+logger = structlog.get_logger()
+
+
+class RetryTrackerMiddleware(TaskiqMiddleware):
+    def pre_execute(self, message: TaskiqMessage) -> TaskiqMessage:
+        """Відстежує, чи є цей запуск повторним (retry)."""
+        # Taskiq зберігає кількість спроб у labels
+        retry_count = message.labels.get("retry_count", 0)
+
+        if int(retry_count) > 0:
+            logger.warning(
+                "task_retry_attempt",
+                task_id=message.task_id,
+                task_name=message.task_name,
+                attempt=retry_count,
+            )
+        return message