Отмена задачи сельдерея изнутри самого себя?

Question

Я знаю, что мог return, но мне интересно, есть ли что-то еще, особенно для вспомогательных методов, где задача, в которой return None заставит вызывающего добавить проверку шаблонов при каждом вызове.

Я нашел InvalidTaskError, но никакой реальной документации - это внутренняя вещь? Уместно ли это поднять?

Я искал что-то вроде self.abort(), похожее на self.retry(), но ничего не видел.

Вот пример, где я буду использовать его.

def helper(task, arg): 
    if unrecoverable_problems(arg): 
     # abort the task 
     raise InvalidTaskError() 

@task(bind=True) 
task_a(self, arg): 
    helper(task=self, arg=arg) 
    do_a(arg) 

@task(bind=True) 
task_b(self, arg): 
    helper(task=self, arg=arg) 
    do_b(arg) 

Answer 1

После того, как вы сделали больше копания, я нашел пример, используя Reject;

(скопировано с доком страницы)

Задача может поднять Отклонить, чтобы отклонить сообщение о задаче AMQPs метода basic_reject. Это не будет иметь никакого эффекта, если Task.acks_late не включен.

Отказ от сообщения имеет тот же эффект, что и его подделка, но некоторые брокеры могут реализовать дополнительные функции, которые могут быть использованы. Например, RabbitMQ поддерживает концепцию Мертвых буквенных обменов, где очередь может быть настроена на использование обмена мертвой буквой, на который отклоняются сообщения, отправленные по почте.

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

Примера использование отвергает, когда задача вызывает отказ от состояния памяти:

import errno 
from celery.exceptions import Reject 

@app.task(bind=True, acks_late=True) 
def render_scene(self, path): 
    file = get_file(path) 
    try: 
     renderer.render_scene(file) 

    # if the file is too big to fit in memory 
    # we reject it so that it's redelivered to the dead letter exchange 
    # and we can manually inspect the situation. 
    except MemoryError as exc: 
     raise Reject(exc, requeue=False) 
    except OSError as exc: 
     if exc.errno == errno.ENOMEM: 
      raise Reject(exc, requeue=False) 

    # For any other error we retry after 10 seconds. 
    except Exception as exc: 
     raise self.retry(exc, countdown=10)