Canvas should not use current_app, instead take the app from the first task. Closes celery#1249

Author: ask

celery/canvas.py

@@ -19,7 +19,6 @@
 from kombu.utils import cached_property, fxrange, kwdict, reprcall, uuid
 
 from celery import current_app
-from celery.local import Proxy
 from celery.utils.compat import chain_from_iterable
 from celery.result import AsyncResult, GroupResult
 from celery.utils.functional import (
@@ -28,8 +27,6 @@
 )
 from celery.utils.text import truncate
 
-Chord = Proxy(lambda: current_app.tasks['celery.chord'])
-
 
 class _getitem_property(object):
     """Attribute -> dict key descriptor.
@@ -278,7 +275,8 @@ def __init__(self, *tasks, **options):
         self.subtask_type = 'chain'
 
     def __call__(self, *args, **kwargs):
-        return self.apply_async(args, kwargs)
+        if self.tasks:
+            return self.apply_async(args, kwargs)
 
     @classmethod
     def from_dict(self, d):
@@ -288,6 +286,10 @@ def from_dict(self, d):
             tasks[0]['args'] = d['args'] + tasks[0]['args']
         return chain(*d['kwargs']['tasks'], **kwdict(d['options']))
 
+    @property
+    def type(self):
+        return self._type or self.tasks[0].type
+
     def __repr__(self):
         return ' | '.join(map(repr, self.tasks))
 Signature.register_type(chain)
@@ -394,10 +396,16 @@ def from_dict(self, d):
         return group(tasks, **kwdict(d['options']))
 
     def __call__(self, *partial_args, **options):
-        tasks, result, gid, args = self.type.prepare(
-            options, map(Signature.clone, self.tasks), partial_args,
-        )
-        return self.type(tasks, result, gid, args)
+        tasks = [task.clone() for task in self.tasks]
+        if not tasks:
+            return
+        # taking the app from the first task in the list,
+        # there may be a better solution to this, e.g.
+        # consolidate tasks with the same app and apply them in
+        # batches.
+        type = tasks[0].type.app.tasks[self['task']]
+        tasks, result, gid, args = type.prepare(options, tasks, partial_args)
+        return type(*type.prepare(options, tasks, partial_args))
 
     def _freeze(self, _id=None):
         opts = self.options
@@ -428,7 +436,6 @@ def __repr__(self):
 
 
 class chord(Signature):
-    Chord = Chord
 
     def __init__(self, header, body=None, task='celery.chord',
                  args=(), kwargs={}, **options):
@@ -450,8 +457,12 @@ def _unpack_args(header=None, body=None, **kwargs):
         # than manually popping things off.
         return (header, body), kwargs
 
+    @property
+    def type(self):
+        return self._type or self.tasks[0].type
+
     def __call__(self, body=None, **kwargs):
-        _chord = self.Chord
+        _chord = self.type.app.tasks['celery.chord']
         body = (body or self.kwargs['body']).clone()
         kwargs = dict(self.kwargs, body=body, **kwargs)
         if _chord.app.conf.CELERY_ALWAYS_EAGER:
@@ -481,13 +492,8 @@ def __repr__(self):
             return self.body.reprcall(self.tasks)
         return '<chord without body: %r>' % (self.tasks, )
 
-    @property
-    def tasks(self):
-        return self.kwargs['header']
-
-    @property
-    def body(self):
-        return self.kwargs.get('body')
+    tasks = _getitem_property('kwargs.header')
+    body = _getitem_property('kwargs.body')
 Signature.register_type(chord)