如何在 Celery 中创建一系列条件子任务？

Question

我正在创建一个应用程序，该应用程序将创建要执行的任务链，但该链将根据用户对添加该部分的需求执行任务。

例如，如果用户想要start_boo，则链可能是：

def start_boo():
    chain = start_foo.s() | start_bar.s() | start_baz.s()
    chain()

但是，如果foo 和baz 已经启动，我们不想这样做；而是更喜欢类似的东西：

def start_boo(foo=True, bar=True, baz=True):
    if not (foo or bar or baz):
        raise Exception("At least one should be true...")
    chain = None
    if foo:
       chain |= start_foo.s()
    if bar:
        chain |= start_bar.s()
    if baz:
        chain |= start_baz.s()
    chain()

start_boo(foo=False, baz=False)

但是，由于各种原因，这不起作用。

有这样的成语吗？

Answer 1

习语是来自functools 的reduce 函数。您可以执行以下操作：

def start_boo(foo=True, bar=True, baz=True):
    if not (foo or bar or baz):
        raise Exception("At least one should be true...")

    todo_tasks = [foo, bar, baz]
    start_tasks = [start_foo, start_bar, start_baz]

    # tasks contains start tasks which should be done per the options.
    # if it's False in todo_tasks, its associated start_task isn't added
    tasks = [start_task for todo, start_task in zip(todo_tasks, start_tasks) if todo]
    first_task, rest = *tasks  

    # start with the first task to be completed and chain it with remaining tasks
    chain = functools.reduce(lambda x, y: x | y.s(), rest, first_task.s())
    chain()