Откат к нормальной функции, если сельдерей не активен

Question

Мне требуется простой способ взлома для синхронного запуска функции, если сельдерей не активен.

Я пытался: is_celery_working возвращает False, хотя и сельдерей, и Redis оба работают (запускали celery -A project worker -l debug и redis-server соответственно).Также get_celery_worker_status всегда дает error в статусе.

Я использую сельдерей с Джанго.

from project.celery import app

def is_celery_working():
    result = app.control.broadcast('ping', reply=True, limit=1)
    return bool(result)  # True if at least one result


def sync_async(func):
    if is_celery_working():
        return func.delay
    else:
        return func

sync_async(some_func)(**its_args, **its_kwrgs)

def get_celery_worker_status():
    error_key = 'error'

    try:
        from celery.task.control import inspect
        insp = inspect()
        d = insp.stats()
        if not d:
            d = {error_key: 'No running Celery workers were found.'}
    except IOError as e:
        from errno import errorcode
        msg = "Error connecting to the backend: " + str(e)
        if len(e.args) > 0 and errorcode.get(e.args[0]) == 'ECONNREFUSED':
            msg += ' Check that the RabbitMQ server is running.'
        d = {error_key: msg}
    except ImportError as e:
        d = {error_key: str(e)}
    return d

def sync_async(func):
    status = get_celery_worker_status()
    if 'error' not in status:
        return func.delay
    else:
        return func

sync_async(some_func)(**its_args, **its_kwrgs)

Answer 1

def sync_async(func, *args, **kwargs):
    try:
        func.delay(*args, **kwargs)
    except Exception as error:
        print('Celery not active', error)
        func(*args, **kwargs)

Это просто выдает ошибку, если сервер Redis не работает.Это работало хорошо для меня, так как я предполагаю, что если Redis не работает, то сельдерей останавливается.

Answer 2

Ваша простая is_celery_working функция выглядит правильно.Если вы получаете False, вы можете увеличить время ожидания до 5 или 10 секунд, используя необязательный параметр timeout.

def is_celery_working():
    result = app.control.broadcast('ping', reply=True, limit=1, timeout=5.0)
    return bool(result)  # True if at least one result