Сельдерей: как проигнорировать задачу в аккорде или цепочке?

Я использую сельдерей, у меня есть несколько задач, которые нужно выполнить в порядке.

Например, у меня есть эта задача:

@celery.task
def tprint(word):
    print word

И я хочу сделать что-то вроде этого:

>>> chain(tprint.s('a') | tprint.s('b'))()

Тогда я получаю TypeError: tprint() takes exactly 1 argument (2 given).

То же самое с аккордом, в этой ситуации, в которой мне нужна задача, которая будет выполнена после группы задач:

>>> chord([tprint.s('a'), tprint.s('b')])(tprint.s('c'))

Итак, как справиться с этой ситуацией? Меня не волнует результат каждой задачи, но их нужно выполнить в порядке.

Добавить второй параметр не будет работать:

@celery.task
def tprint(word, ignore=None):
    print word

>>> chain(tprint.s('a', 0) | tprint.s('b'))()

Это напечатает "a" и "None".

Ответ 1

Существует встроенная функция, чтобы игнорировать результат в цепочке, а другие - неизменяемую подзадачу. Вы можете использовать ярлык .si() вместо .s() или .subtask(immutable = True)

Подробнее здесь: http://docs.celeryproject.org/en/master/userguide/canvas.html#immutability

Ответ 2

Одно из возможных решений уже опубликовано, но я хотел бы добавить дополнительное разъяснение и альтернативное решение (а в некоторых случаях и более высокое).

Ошибка, которую вы видите, которая указывает, что ваша подпись задачи должна учитывать второй параметр, связана с тем, что при вызове задач в chain, сельдерей автоматически подталкивает каждую задачу result как первый параметр следующей задачи.

Из docs:

Задачи могут быть связаны между собой, что на практике означает добавление задачи обратного вызова:

>>> res = add.apply_async((2, 2), link=mul.s(16))
>>> res.get()
4

Связанная задача будет применяться с результатом ее родительской задачи как первого аргумента

Поэтому в вашем случае вы можете переписать свою задачу следующим образом:

@celery.task
def tprint(result, word):
    print word

Если вы не собираетесь ничего делать с результатом, вы можете также игнорировать его, меняя декоратор таким образом

@celery.task(ignore_result=True)

И тогда вам не придется менять свою подпись задачи.Забастовкa >

Извините, последняя точка нуждается в дальнейших исследованиях.

Ответ 3

Вы можете попробовать сделать что-то вроде этого. Вместо того, чтобы иметь один параметр для функции tprint, вы можете иметь 2 параметра

def tprint(word, x=None):
    print word

то

chain(tprint.s('a', 0) | tprint.s('b'))()

Ответ 4

Наконец, найдите обходной путь для этого, декоратор цепи выполнит эту работу.

Я не знаю, как именно это сделал сельдерей, но сельдерей кажется, что сила привязывает предыдущий результат задачи к первому аргументу следующей задачи.

Итак, вот пример:

def chain_deco(func):

    @functools.wraps(func)
    def wrapper(chain=None, *args, **kwargs):
        if chain is False:
            return False

        func(*args, **kwargs)
        return True

    return wrapper


@celery.task
@chain_deco
def hello(word):
    print "hello %s" % word

Теперь это даст правильный результат.

>>> (hello.s(word='a') | hello.s(word='b'))()

ИЛИ

>>> (hello.s('a') | hello.s('b'))(True)

И декоратор также предоставляет возможность остановить цепочку посередине (сделать более поздний каскад неудачным.)

Тот же механизм должен работать и для chord.