Ответы, комментарии, мнения на вопрос.

Да вопрос очень интересный давайте обсудим и вместе найдем ответ кто что знает или думает?

Знаете ответ на этот вопрос? Опубликуйте его ваше мнение будет интересно другим пользователям!

2 комментария на ««Как прокинуть логи таски celery в роутер fastapi?»»

1. 

   Ответ:

   Ховокмюдл

   Тут могла бы быть ваша ссылка подробнее в личном кабинете.

   26 сентября, 2024

   Для того чтобы прокинуть логи заданий Celery в роутер FastAPI, можно использовать стандартные инструменты логирования в Python, такие как модуль logging.

   Во-первых, необходимо настроить логирование в Celery. Для этого можно использовать параметр `task_track_started` в конфигурации Celery, чтобы отслеживать статус выполнения задания. Например:

   “`python
   app.conf.update(
   task_track_started=True,
   )
   “`

   Затем, можно использовать модуль logging в FastAPI для перенаправления логов Celery в нужный поток вывода. Для этого нужно настроить логгер с нужными обработчиками и форматтерами. Например:

   “`python
   import logging

   logger = logging.getLogger(‘celery’)
   logger.addHandler(logging.StreamHandler())
   logger.setLevel(logging.INFO)
   “`

   После этого, все логи заданий Celery будут выводиться в поток вывода, который был настроен в логгере. Таким образом, вы сможете видеть логи заданий Celery в роутере FastAPI.

   Ответить
2. 

   Ответ:

   Nanatlat

   Тут могла бы быть ваша ссылка подробнее в личном кабинете.

   1 октября, 2024

   Для того чтобы прокинуть логи таски celery в роутер fastapi, можно воспользоваться библиотекой celery.events и celery log handlers. Можно настроить celery для того, чтобы отправлять логи в стандартный python logger с помощью celery log handlers, а затем создать роутер в fastapi, который будет отображать эти логи.

   Пример:

   “`
   import logging
   from celery import Celery
   from celery.signals import task_failure, task_success
   from fastapi import FastAPI

   app = FastAPI()

   # Создаем celery приложение
   celery = Celery(‘tasks’, broker=’redis://localhost’)

   # Настраиваем логирование celery
   logger = logging.getLogger(__name__)
   handler = logging.StreamHandler()
   handler.setLevel(logging.INFO)
   logger.addHandler(handler)
   celery.log.setup_logger(logger)

   # Роутер для отображения логов
   @app.get(‘/logs’)
   def get_logs():
   # Здесь можно вернуть логи в каком-то формате, например из файла
   pass

   # Обработчики для событий тасок
   @task_failure.connect
   def task_failed_handler(sender=None, task_id=None, exception=None, args=None, kwargs=None, traceback=None, einfo=None, **other):
   logger.error(f’Task {sender} with id {task_id} failed with exception: {exception}’)

   @task_success.connect
   def task_success_handler(sender=None, task_id=None, result=None, **other):
   logger.info(f’Task {sender} with id {task_id} was successful with result: {result}’)
   “`

   Таким образом, можно настроить celery для отправки логов в стандартный python logger и создать роутер в fastapi для отображения этих логов.

   Ответить