celery - 如何从任务内部使 celery 任务失败？

Question

在某些情况下，我想让 celery 任务从该任务中失败。我尝试了以下方法:

from celery.task import task
from celery import states

@task()
def run_simulation():
    if some_condition:
        run_simulation.update_state(state=states.FAILURE)
        return False

但是，任务仍然报告已成功:

Task sim.tasks.run_simulation[9235e3a7-c6d2-4219-bbc7-acf65c816e65] succeeded in 1.17847704887s: False

似乎只能在任务运行时修改状态，并且一旦完成 - celery 会将状态更改为它认为是结果的任何内容(请参阅 this question )。有什么方法可以在不通过引发异常而使任务失败的情况下让 celery 返回任务失败？

Answer 1

要将任务标记为失败而不引发异常，请将任务状态更新为 FAILURE然后提出一个 Ignore异常，因为返回任何值都会将任务记录为成功，例如:

from celery import Celery, states
from celery.exceptions import Ignore

app = Celery('tasks', broker='amqp://guest@localhost//')

@app.task(bind=True)
def run_simulation(self):
    if some_condition:
        # manually update the task state
        self.update_state(
            state = states.FAILURE,
            meta = 'REASON FOR FAILURE'
        )

        # ignore the task so no other state is recorded
        raise Ignore()

但最好的方法是从您的任务中引发异常，您可以创建自定义异常来跟踪这些失败:

class TaskFailure(Exception):
   pass

并从您的任务中引发此异常:

if some_condition:
    raise TaskFailure('Failure reason')