Consistent decoding of exception result

celery/backends/amqp.py

@@ -256,14 +256,12 @@ class AMQPBackend(BaseBackend):
             results = deque()
             push_result = results.append
             push_cache = self._cache.__setitem__
-            to_exception = self.exception_to_python
+            decode_result = self.decode_result
 
             def on_message(message):
-                body = message.decode()
+                body = decode_result(message.body)
                 state, uid = getfields(body)
                 if state in READY_STATES:
-                    if state in PROPAGATE_STATES:
-                        body['result'] = to_exception(body['result'])
                     push_result(body) \
                         if uid in task_ids else push_cache(uid, body)
 

celery/backends/base.py

@@ -181,6 +181,12 @@ class BaseBackend(object):
         _, _, payload = dumps(data, serializer=self.serializer)
         return payload
 
+    def decode_result(self, payload):
+        meta = self.decode(payload)
+        if meta['status'] in self.EXCEPTION_STATES:
+            meta['result'] = self.exception_to_python(meta['result'])
+        return meta
+
     def decode(self, payload):
         payload = PY3 and payload or str(payload)
         return loads(payload,
@@ -264,11 +270,7 @@ class BaseBackend(object):
 
     def get_result(self, task_id):
         """Get the result of a task."""
-        meta = self.get_task_meta(task_id)
-        if meta['status'] in self.EXCEPTION_STATES:
-            return self.exception_to_python(meta['result'])
-        else:
-            return meta['result']
+        return self.get_task_meta(task_id).get('result')
 
     def get_children(self, task_id):
         """Get the list of subtasks sent by a task."""
@@ -436,13 +438,13 @@ class KeyValueStoreBackend(BaseBackend):
         if hasattr(values, 'items'):
             # client returns dict so mapping preserved.
             return {
-                self._strip_prefix(k): self.decode(v)
+                self._strip_prefix(k): self.decode_result(v)
                 for k, v in items(values) if v is not None
             }
         else:
             # client returns list so need to recreate mapping.
             return {
-                bytes_to_str(keys[i]): self.decode(value)
+                bytes_to_str(keys[i]): self.decode_result(value)
                 for i, value in enumerate(values) if value is not None
             }
 
@@ -500,7 +502,7 @@ class KeyValueStoreBackend(BaseBackend):
         meta = self.get(self.get_key_for_task(task_id))
         if not meta:
             return {'status': states.PENDING, 'result': None}
-        return self.decode(meta)
+        return self.decode_result(meta)
 
     def _restore_group(self, group_id):
         """Get task metadata for a task by id."""

celery/backends/redis.py

@@ -221,7 +221,7 @@ class RedisBackend(KeyValueStoreBackend):
             callback = maybe_signature(request.chord, app=app)
             total = callback['chord_size'] + totaldiff
             if readycount == total:
-                decode, unpack = self.decode, self._unpack_chord_result
+                decode, unpack = self.decode_result, self._unpack_chord_result
                 resl, _, _ = client.pipeline()  \
                     .lrange(jkey, 0, total)     \
                     .delete(jkey)               \

celery/canvas.py

@@ -618,7 +618,8 @@ class group(Signature):
                         task.args = tuple(partial_args) + tuple(task.args)
                     yield task, task.freeze(group_id=group_id, root_id=root_id)
 
-    def _apply_tasks(self, tasks, producer=None, app=None, **options):
+    def _apply_tasks(self, tasks, producer=None, app=None,
+                     add_to_parent=None, **options):
         app = app or self.app
         with app.producer_or_acquire(producer) as producer:
             for sig, res in tasks:

celery/result.py

@@ -174,9 +174,7 @@ class AsyncResult(ResultBase):
             self._maybe_set_cache(meta)
             status = meta['status']
             if status in PROPAGATE_STATES and propagate:
-                raise self.backend.exception_to_python(meta['result'])
-            if status in EXCEPTION_STATES:
-                return self.backend.exception_to_python(meta['result'])
+                raise meta['result']
             return meta['result']
     wait = get  # deprecated alias to :meth:`get`.
 
@@ -345,8 +343,6 @@ class AsyncResult(ResultBase):
 
     def _set_cache(self, d):
         state, children = d['status'], d.get('children')
-        if state in states.EXCEPTION_STATES:
-            d['result'] = self.backend.exception_to_python(d['result'])
         if children:
             d['children'] = [
                 result_from_tuple(child, self.app) for child in children

docs/getting-started/next-steps.rst

@@ -317,7 +317,7 @@ exception, in fact ``result.get()`` will propagate any errors by default::
     File "/opt/devel/celery/celery/result.py", line 113, in get
         interval=interval)
     File "/opt/devel/celery/celery/backends/amqp.py", line 138, in wait_for
-        raise self.exception_to_python(meta['result'])
+        raise meta['result']
     TypeError: add() takes exactly 2 arguments (1 given)
 
 If you don't wish for the errors to propagate then you can disable that

docs/userguide/canvas.rst

@@ -735,7 +735,7 @@ to the :exc:`~@ChordError` exception:
       File "*/celery/result.py", line 120, in get
         interval=interval)
       File "*/celery/backends/amqp.py", line 150, in wait_for
-        raise self.exception_to_python(meta['result'])
+        raise meta['result']
     celery.exceptions.ChordError: Dependency 97de6f3f-ea67-4517-a21c-d867c61fcb47
         raised ValueError('something something',)