最新消息:雨落星辰是一个专注网站SEO优化、网站SEO诊断、搜索引擎研究、网络营销推广、网站策划运营及站长类的自媒体原创博客

python - Getting the correct logger with Celery in all circumstances: outside and inside worker celery processes and tasks - Sta

programmeradmin2浏览0评论

We use Celery with some of our applications but not for all. I am trying to devise a get_logger function that would retrieve a proper logger object in all circumstances. It should work in all of the following situations:

  1. Celery cannot be imported.
  2. Celery can be imported, but the function is not called from a celery task or worker.
  3. The function is called from a celery worker but not a task.
  4. Te function is called from a celery task.

Additionally, I would like the get_logger to accept the same arguments as logging.getLogger. Namely, it should accept name=None.

import logging
from typing import Optional

try:
    import celery
    import celery.app
    from celery.utils.log import base_logger, task_logger
    from celery.utils.log import get_logger as celery_get_logger
    from celery.utils.log import get_task_logger as celery_get_task_logger

except ModuleNotFoundError:
    celery = None


def get_logger(name: Optional[str] = None) -> logging.Logger:
    """Get the logger object."""

    # Option 1 & 2: If celery is not installed or we are not in a celery worker.
    if celery is None or celery.current_app == celery.app.default_app:
        return logging.getLogger(name=name)

    # Option 3: Get a logger from celery if we are not in a celery task.
    if not celery.current_task or celery.current_task.request.id is None:
        return base_logger if name is None else celery_get_logger(name=name)

    # Option 4: Get the celery task logger.
    return task_logger if name is None else celery_get_task_logger(name=name)

Option 2 is not working. I found that it is possible to tell if the call is made from a celery worker by checking the name of the process assuming it has been set at the worker process startup. But that would depend on setting the worker process name in the applications.

Question: is there a way to tell if we are in a celery worker process without inspecting the process' name? Is there a variable set maybe, like celery.current_task just for the worker?

发布评论

评论列表(0)

  1. 暂无评论