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

admin管理员组

文章数量:1414998

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?

本文标签：