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:
- Celery cannot be imported.
- Celery can be imported, but the function is not called from a celery task or worker.
- The function is called from a celery worker but not a task.
- 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?
发布者：admin，转转请注明出处：http://www.yc00.com/questions/1744755635a4591863.html
评论列表（0条）