Целевая цепочка сельдерея и доступ к ** kwargs

У меня есть ситуация, аналогичная описанной здесь, за исключением того, что вместо цепочки задач с несколькими аргументами я хочу связать задачи, которые возвращают словарь с несколькими записями,

Это - очень свободно и абстрактно --- что я пытаюсь сделать:

tasks.py

@task()
def task1(item1=None, item2=None):
  item3 = #do some stuff with item1 and item2 to yield item3
  return_object = dict(item1=item1, item2=item2, item3=item3)
  return return_object

def task2(item1=None, item2=None, item3=None):
  item4 = #do something with item1, item2, item3 to yield item4
  return_object = dict(item1=item1, item2=item2, item3=item3, item4=item4)
  return return_object

Работа с ipython, я могу вызвать task1 индивидуально и асинхронно, без проблем.

Я также могу вызвать task2 отдельно с результатом, возвращаемым task1 в качестве аргумента с двойной звездой:

>>res1 = task1.s(item1=something, item2=something_else).apply_async()
>>res1.status
'SUCCESS'
>>res2 = task2.s(**res1.result).apply_async()
>>res2.status
'SUCCESS

Однако то, что я в конечном итоге хочу достичь, это тот же конечный результат, что и выше, но через цепочку, и здесь я не могу понять, как создать экземпляр task2 не с (позиционными) аргументами, возвращаемыми task1, а с task1.result как ** kwargs:

chain_result = (task1.s(item1=something, item2=something_else) | task2.s()).apply_async()  #THIS DOESN'T WORK!

Я подозреваю, что могу вернуться и переписать свои задачи так, чтобы они возвращали позиционные аргументы вместо словаря, и это может прояснить ситуацию, но мне кажется, что должен быть какой-то способ доступа к возвращенному объекту task1 в task2 с эквивалентной функциональностью ** двойной звезды. Я также подозреваю, что я пропустил кое-что здесь довольно очевидное о реализации подзадачи Celery или * args vs. ** kwargs.

Надеюсь, это имеет смысл. И спасибо заранее за любые советы.

Ответ 1

chain, а остальные примитивы холста находятся в семействе функциональные утилиты, такие как map и reduce.

например. где map(target, items) вызывает target(item) для каждого элемента в списке, Python имеет редко используемую версию карты под названием itertools.starmap, который вместо этого вызывает target(*item).

Пока мы могли добавить starchain и даже kwstarchain в панель инструментов, эти будет очень специализированным и, вероятно, не будет использоваться так часто.

Интересно, что Python сделал ненужным с помощью выражений списка и генератора, так что карта заменяется на [target(item) for item in item] и starmap с помощью [target(*item) for item in item].

Поэтому вместо того, чтобы внедрять несколько альтернатив для каждого примитива, я думаю, что мы должны сосредоточиться на поиске более гибкого способа поддержки этого, например. например, с использованием выражений генератора на основе сельдерея (если возможно, и если не что-то подобное)

Ответ 2

Это мое решение проблемы, используя абстрактный класс задачи:

from __future__ import absolute_import
from celery import Task
from myapp.tasks.celery import app   


class ChainedTask(Task):
    abstract = True    

    def __call__(self, *args, **kwargs):
        if len(args) == 1 and isinstance(args[0], dict):
            kwargs.update(args[0])
            args = ()
        return super(ChainedTask, self).__call__(*args, **kwargs)

@app.task(base=ChainedTask)
def task1(x, y):
    return {'x': x * 2, 'y': y * 2, 'z': x * y}    


@app.task(base=ChainedTask)
def task2(x, y, z):
    return {'x': x * 3, 'y': y * 3, 'z': z * 2}

Теперь вы можете определить и выполнить свою цепочку как таковую:

from celery import chain

pipe = chain(task1.s(x=1, y=2) | task2.s())
pipe.apply_async()

Ответ 3

Так как это не встроено в сельдерей, я написал функцию декоратора к чему-то подобному себе.

# Use this wrapper with functions in chains that return a tuple. The
# next function in the chain will get called with that the contents of
# tuple as (first) positional args, rather than just as just the first
# arg. Note that both the sending and receiving function must have
# this wrapper, which goes between the @task decorator and the
# function definition. This wrapper should not otherwise interfere
# when these conditions are not met.

class UnwrapMe(object):
    def __init__(self, contents):
        self.contents = contents

    def __call__(self):
        return self.contents

def wrap_for_chain(f):
    """ Too much deep magic. """
    @functools.wraps(f)
    def _wrapper(*args, **kwargs):
        if type(args[0]) == UnwrapMe:
            args = list(args[0]()) + list(args[1:])
        result = f(*args, **kwargs)

        if type(result) == tuple and current_task.request.callbacks:
            return UnwrapMe(result)
        else:
            return result
    return _wrapper

Mine распаковывается как концепция starchain, но вы можете легко изменить ее, чтобы развернуть kwargs.