Explorar o código

Support for piping a subtask to a chain

This allows to do things like:
pipe = sometask.s() | someothertask.s()
new_pipe = mytask.s() | pipe
Steeve Morin %!s(int64=12) %!d(string=hai) anos
pai
achega
79d04e3de0
Modificáronse 1 ficheiros con 3 adicións e 1 borrados
  1. 3 1
      celery/canvas.py

+ 3 - 1
celery/canvas.py

@@ -166,7 +166,9 @@ class Signature(dict):
                     for link in maybe_list(self.options.get('link')) or []))))
 
     def __or__(self, other):
-        if isinstance(other, chain):
+        if not isinstance(self, chain) and isinstance(other, chain):
+            return chain((self,) + other.tasks)
+        elif isinstance(other, chain):
             return chain(*self.tasks + other.tasks)
         elif isinstance(other, Signature):
             if isinstance(self, chain):