Jak zaimplementować autoregulację dla zadań związanych z selerem

Question

W programie Seler można wykonać dowolne zadanie w przypadku wyjątku. Możesz to zrobić tak:

@task(max_retries=5) 
def div(a, b): 
    try: 
     return a/b 
    except ZeroDivisionError, exc: 
     raise div.retry(exc=exc) 

W tym przypadku, jeśli chcesz podzielić przez zero, zadanie zostanie powtórzone pięć razy. Ale musisz sprawdzić błędy w kodzie jawnie. Zadanie nie zostanie wznowione, jeśli pominiesz blok .

Chcę moje funkcje wyglądać:

@celery.task(autoretry_on=ZeroDivisionError, max_retries=5) 
def div(a, b): 
    return a/b 

Answer 1

Szukałem tego problemu na chwilę, ale występuje tylko this feature request.

zdecyduję się napisać własny dekorator do prowadzenia automatycznych ponownych prób:

def task_autoretry(*args_task, **kwargs_task): 
    def real_decorator(func): 
     @task(*args_task, **kwargs_task) 
     @functools.wraps(func) 
     def wrapper(*args, **kwargs): 
      try: 
       func(*args, **kwargs) 
      except kwargs_task.get('autoretry_on', Exception), exc: 
       wrapper.retry(exc=exc) 
     return wrapper 
    return real_decorator 

Z tego dekoratora mogę rewriter mój poprzedni zadanie:

@task_autoretry(autoretry_on=ZeroDivisionError, max_retries=5) 
def div(a, b): 
    return a/b 

Answer 2

Mam zmodyfikowany your answer do pracy z istniejący interfejs API selera (obecnie 3.1.17)

class MyCelery(Celery): 
    def task(self, *args_task, **opts_task): 
     def real_decorator(func): 
      sup = super(MyCelery, self).task 

      @sup(*args_task, **opts_task) 
      @functools.wraps(func) 
      def wrapper(*args, **kwargs): 
       try: 
        func(*args, **kwargs) 
       except opts_task.get('autoretry_on', Exception) as exc: 
        logger.info('Yo! We did it!') 
        wrapper.retry(exc=exc, args=args, kwargs=kwargs) 
      return wrapper 
     return real_decorator 

Następnie, w swoich zadaniach

app = MyCelery() 
app.config_from_object('django.conf:settings') 
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS) 

@app.task(autoretry_on=Exception) 
def mytask(): 
    raise Exception('Retrying!') 

Umożliwia to dodanie funkcji autoretry_on do zadań bez konieczności używania oddzielnego dekoratora do definiowania zadań.

Answer 3

Seler (od wersji 4.0) ma dokładnie to, czego szukaliśmy:

@app.task(autoretry_for=(SomeException,)) 
def my_task(): 
    ... 

Patrz: http://docs.celeryproject.org/en/latest/userguide/tasks.html#automatic-retry-for-known-exceptions