W jaki sposób można uchwycić niestandardowy wyjątek od pracownika Selera lub zatrzymać przedrostkiem `celery.backends.base`?

Question

Moje zadanie Selery podnosi wyjątek niestandardowy NonTransientProcessingError, który następnie zostaje przechwycony przez AsyncResult.get(). Tasks.py:

class NonTransientProcessingError(Exception): 
    pass 

@shared_task() 
def throw_exception(): 
    raise NonTransientProcessingError('Error raised by POC model for test purposes') 

W konsoli Pythona:

from my_app.tasks import * 
r = throw_exception.apply_async() 
try: 
    r.get() 
except NonTransientProcessingError as e: 
    print('caught NonTrans in type specific except clause') 

Ale mój zwyczaj Wyjątkiem jest my_app.tasks.NonTransientProcessingError, natomiast wyjątek podniesiony przez AsyncResult.get() jest celery.backends.base.NonTransientProcessingError, więc moja except klauzula nie .

Traceback (most recent call last): 
    File "<input>", line 4, in <module> 
    File "/...venv/lib/python3.5/site-packages/celery/result.py", line 175, in get 
    raise meta['result'] 
celery.backends.base.NonTransientProcessingError: Error raised by POC model for test purposes 

Jeśli przechwycę wyjątek w ramach zadania, działa poprawnie. Dopiero, gdy wyjątek zostanie podniesiony do wywołania .get(), zostanie zmieniona jego nazwa.

Jak mogę podnieść niestandardowy wyjątek i złapać go poprawnie?

Potwierdziłem, że to samo dzieje się po zdefiniowaniu klasy Task i podniesienia niestandardowego wyjątku w metodzie on_failure. Poniższa działa:

try: 
    r.get() 
except Exception as e: 
    if type(e).__name__ == 'NonTransientProcessingError': 
     print('specific exception identified') 
    else: 
     print('caught generic but not identified') 

Wyjścia:

specific exception identified 

Ale to nie może być najlepszym sposobem osiągnięcia tego celu? Idealnie chciałbym wychwycić klasy wyjątków wyjątków dla kategorii zachowań.

Używam Django 1.8.6, Python 3.5 i Seler 3.1.18, z systemem Redis 3.1.18, Python redis lib 2.10.3.

Answer 1

import celery 
from celery import shared_task 


class NonTransientProcessingError(Exception): 
    pass 


class CeleryTask(celery.Task): 

    def on_failure(self, exc, task_id, args, kwargs, einfo): 
     if isinstance(exc, NonTransientProcessingError): 
      """ 
      deal with NonTransientProcessingError 
      """ 
      pass 

    def run(self, *args, **kwargs): 
     pass 


@shared_task(base=CeleryTask) 
def add(x, y): 
    raise NonTransientProcessingError 

Użyj zadania podstawowego z funkcją zwrotną on_failure, aby przechwycić niestandardowy wyjątek.