芹菜链 - 如果任何任务失败，执行 x，否则 y答案

【问题标题】：Celery chain - if any tasks fail, do x, else y芹菜链 - 如果任何任务失败，执行 x，否则 y
【发布时间】：2022-01-13 01:01:56
【问题描述】：

我刚刚进入我的 Django 项目中的 Celery 链。我有以下功能：

def orchestrate_tasks_for_account(account_id):

    # Get the account, set status to 'SYNC' until the chain is complete
    account = Account.objects.get(id=account_id)
    account.status = "SYNC"
    account.save()

    chain = task1.s(account_id) | task2.s() | task3.s()
    chain()

    # if any of the tasks in the chain failed, set account.status = 'ERROR'
    # else set the account.status = 'OK'

链按预期工作，但我不确定如何从链中获取反馈并根据结果更新帐户

换句话说，如果链中的任何任务失败，我想将帐户状态设置为'ERROR'，否则我想将帐户状态设置为'OK'

我对 Celery 文档关于如何使用 if/else 处理错误感到困惑，就像我在上面最后两行中评论的那样。

有人有这方面的经验吗？

【问题讨论】：

标签： celery celery-task

【解决方案1】：

好的，这就是我的想法

我在此解决方案中利用了 waiting 库

from celery import chain
from waiting import wait


def orchestrate_tasks_for_account(account_id):

    account = Account.objects.get(id=account_id)
    account.status = "SYNC"
    account.save()

    job = chain(
        task1.s(account_id),
        task2.s(),
        task3.s()
        )
    result = job.apply_async()

    wait(
        lambda: result.ready(), # when async job is completed...
        timeout_seconds=1800, # wait 1800 seconds (30 minutes)
        waiting_for="task orchestration to complete"
        )

    if result.successful():
        account.status = 'OK'
    else:
        account.status = 'ERROR'

    account.save()

我愿意接受建议以使这变得更好！

【讨论】：