|
@@ -14,6 +14,7 @@ from __future__ import absolute_import
|
|
from .. import current_app
|
|
from .. import current_app
|
|
from ..app import app_or_default, current_task as _current_task
|
|
from ..app import app_or_default, current_task as _current_task
|
|
from ..local import Proxy
|
|
from ..local import Proxy
|
|
|
|
+from ..utils import uuid
|
|
|
|
|
|
from .base import BaseTask, Task, PeriodicTask # noqa
|
|
from .base import BaseTask, Task, PeriodicTask # noqa
|
|
from .sets import group, TaskSet, subtask # noqa
|
|
from .sets import group, TaskSet, subtask # noqa
|
|
@@ -93,7 +94,22 @@ def periodic_task(*args, **options):
|
|
backend_cleanup = Proxy(lambda: current_app.tasks["celery.backend_cleanup"])
|
|
backend_cleanup = Proxy(lambda: current_app.tasks["celery.backend_cleanup"])
|
|
|
|
|
|
|
|
|
|
-def chain(*tasks):
|
|
|
|
- tasks = [task.clone() for task in tasks]
|
|
|
|
- reduce(lambda a, b: a.link(b), tasks)
|
|
|
|
- return tasks[0]
|
|
|
|
|
|
+class chain(object):
|
|
|
|
+
|
|
|
|
+ def __init__(self, *tasks):
|
|
|
|
+ self.tasks = tasks
|
|
|
|
+
|
|
|
|
+ def apply_async(self, **kwargs):
|
|
|
|
+ tasks = [task.clone(task_id=uuid(), **kwargs)
|
|
|
|
+ for task in self.tasks]
|
|
|
|
+ reduce(lambda a, b: a.link(b), tasks)
|
|
|
|
+ tasks[0].apply_async()
|
|
|
|
+ results = [task.type.AsyncResult(task.options["task_id"])
|
|
|
|
+ for task in tasks]
|
|
|
|
+
|
|
|
|
+ def update_parent(result, parent):
|
|
|
|
+ result.parent = parent
|
|
|
|
+ return parent
|
|
|
|
+
|
|
|
|
+ reduce(update_parent, reversed(results))
|
|
|
|
+ return results[-1]
|