如何在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

該成語是reduce函數從functools。您可以執行以下操作：

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()