Cách thực hiện ghi chú tự động cho các nhiệm vụ Celery

Question

Trong cần tây, bạn có thể retry mọi tác vụ trong trường hợp ngoại lệ. Bạn có thể làm như vậy:

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

Trong trường hợp này, nếu bạn muốn chia cho 0, nhiệm vụ sẽ được thử lại năm lần. Nhưng bạn phải kiểm tra lỗi trong mã số rõ ràng là. Tác vụ sẽ không được thử lại nếu bạn bỏ qua khối try-except.

Tôi muốn các chức năng của tôi để trông giống như:

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

Answer 1

Tôi đã tìm kiếm vấn đề này trong một thời gian, nhưng chỉ this feature request được tìm thấy.

tôi quyết định viết trang trí của riêng tôi để thực hiện tự động thử lại:

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 

Với trang trí này tôi có thể Rewriter nhiệm vụ trước đây của tôi:

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

Answer 2

Tôi đã sửa đổi your answer để làm việc với API Celery hiện tại (hiện tại là 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 

Sau đó, trong công việc của bạn

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!') 

Điều này cho phép bạn thêm chức năng autoretry_on vào công việc của mình mà không phải sử dụng công cụ trang trí riêng để xác định tác vụ.

Answer 3

Cần tây (kể từ phiên bản 4.0) có chính xác những gì bạn đang tìm kiếm:

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

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