Merge branch 'master' of github.com:celery/celery

celery/app/base.py

@@ -15,6 +15,7 @@ import warnings
 from collections import defaultdict, deque
 from copy import deepcopy
 from operator import attrgetter
+from functools import wraps
 
 from amqp import promise
 try:
@@ -291,6 +292,20 @@ class Celery(object):
                 '__wrapped__': run}, **options))()
             self._tasks[task.name] = task
             task.bind(self)  # connects task to this app
+
+            autoretry_for = tuple(options.get('autoretry_for', ()))
+            retry_kwargs = options.get('retry_kwargs', {})
+
+            if autoretry_for and not hasattr(task, '_orig_run'):
+
+                @wraps(task.run)
+                def run(*args, **kwargs):
+                    try:
+                        return task._orig_run(*args, **kwargs)
+                    except autoretry_for as exc:
+                        raise task.retry(exc=exc, **retry_kwargs)
+
+                task._orig_run, task.run = task.run, run
         else:
             task = self._tasks[name]
         return task

celery/tests/tasks/test_tasks.py

@@ -100,6 +100,20 @@ class TasksCase(AppCase):
                     raise self.retry(countdown=0, exc=exc)
         self.retry_task_customexc = retry_task_customexc
 
+        @self.app.task(bind=True, autoretry_for=(ZeroDivisionError,),
+                       shared=False)
+        def autoretry_task_no_kwargs(self, a, b):
+            self.iterations += 1
+            return a/b
+        self.autoretry_task_no_kwargs = autoretry_task_no_kwargs
+
+        @self.app.task(bind=True, autoretry_for=(ZeroDivisionError,),
+                       retry_kwargs={'max_retries': 5}, shared=False)
+        def autoretry_task(self, a, b):
+            self.iterations += 1
+            return a/b
+        self.autoretry_task = autoretry_task
+
 
 class MyCustomException(Exception):
     """Random custom exception."""
@@ -209,6 +223,18 @@ class test_task_retries(TasksCase):
             result.get()
         self.assertEqual(self.retry_task.iterations, 2)
 
+    def test_autoretry_no_kwargs(self):
+        self.autoretry_task_no_kwargs.max_retries = 3
+        self.autoretry_task_no_kwargs.iterations = 0
+        self.autoretry_task_no_kwargs.apply((1, 0))
+        self.assertEqual(self.autoretry_task_no_kwargs.iterations, 4)
+
+    def test_autoretry(self):
+        self.autoretry_task.max_retries = 3
+        self.autoretry_task.iterations = 0
+        self.autoretry_task.apply((1, 0))
+        self.assertEqual(self.autoretry_task.iterations, 6)
+
 
 class test_canvas_utils(TasksCase):
 

docs/userguide/tasks.rst

@@ -496,6 +496,45 @@ override this default.
             raise self.retry(exc=exc, countdown=60)  # override the default and
                                                      # retry in 1 minute
 
+Autoretrying
+------------
+
+.. versionadded:: 3.2
+
+Sometimes you may want to retry a task on particular exception. To do so,
+you should wrap a task body with `try-except` statement, for example:
+
+.. code-block:: python
+
+    @app.task
+    def div(a, b):
+        try:
+            return a / b
+        except ZeroDivisionError as exc:
+            raise div.retry(exc=exc)
+
+This may not be acceptable all the time, since you may have a lot of such
+tasks.
+
+Fortunately, you can tell Celery to automatically retry a task using
+`autoretry_for` argument in `~@Celery.task` decorator:
+
+.. code-block:: python
+
+    @app.task(autoretry_for(ZeroDivisionError,))
+    def div(a, b):
+        return a / b
+
+If you want to specify custom arguments for internal `~@Task.retry`
+call, pass `retry_kwargs` argument to `~@Celery.task` decorator:
+
+.. code-block:: python
+
+    @app.task(autoretry_for=(ZeroDivisionError,),
+              retry_kwargs={'max_retries': 5})
+    def div(a, b):
+        return a / b
+
 .. _task-options:
 
 List of Options