665 lines
21 KiB
Python
665 lines
21 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
celery.canvas
|
|
~~~~~~~~~~~~~
|
|
|
|
Composing task workflows.
|
|
|
|
Documentation for some of these types are in :mod:`celery`.
|
|
You should import these from :mod:`celery` and not this module.
|
|
|
|
|
|
"""
|
|
from __future__ import absolute_import
|
|
|
|
from collections import MutableSequence
|
|
from copy import deepcopy
|
|
from functools import partial as _partial, reduce
|
|
from operator import itemgetter
|
|
from itertools import chain as _chain
|
|
|
|
from kombu.utils import cached_property, fxrange, kwdict, reprcall, uuid
|
|
|
|
from celery._state import current_app
|
|
from celery.utils.functional import (
|
|
maybe_list, is_list, regen,
|
|
chunks as _chunks,
|
|
)
|
|
from celery.utils.text import truncate
|
|
|
|
__all__ = ['Signature', 'chain', 'xmap', 'xstarmap', 'chunks',
|
|
'group', 'chord', 'signature', 'maybe_signature']
|
|
|
|
|
|
class _getitem_property(object):
|
|
"""Attribute -> dict key descriptor.
|
|
|
|
The target object must support ``__getitem__``,
|
|
and optionally ``__setitem__``.
|
|
|
|
Example:
|
|
|
|
>>> from collections import defaultdict
|
|
|
|
>>> class Me(dict):
|
|
... deep = defaultdict(dict)
|
|
...
|
|
... foo = _getitem_property('foo')
|
|
... deep_thing = _getitem_property('deep.thing')
|
|
|
|
|
|
>>> me = Me()
|
|
>>> me.foo
|
|
None
|
|
|
|
>>> me.foo = 10
|
|
>>> me.foo
|
|
10
|
|
>>> me['foo']
|
|
10
|
|
|
|
>>> me.deep_thing = 42
|
|
>>> me.deep_thing
|
|
42
|
|
>>> me.deep
|
|
defaultdict(<type 'dict'>, {'thing': 42})
|
|
|
|
"""
|
|
|
|
def __init__(self, keypath):
|
|
path, _, self.key = keypath.rpartition('.')
|
|
self.path = path.split('.') if path else None
|
|
|
|
def _path(self, obj):
|
|
return (reduce(lambda d, k: d[k], [obj] + self.path) if self.path
|
|
else obj)
|
|
|
|
def __get__(self, obj, type=None):
|
|
if obj is None:
|
|
return type
|
|
return self._path(obj).get(self.key)
|
|
|
|
def __set__(self, obj, value):
|
|
self._path(obj)[self.key] = value
|
|
|
|
|
|
def maybe_unroll_group(g):
|
|
"""Unroll group with only one member."""
|
|
# Issue #1656
|
|
try:
|
|
size = len(g.tasks)
|
|
except TypeError:
|
|
try:
|
|
size = g.tasks.__length_hint__()
|
|
except (AttributeError, TypeError):
|
|
pass
|
|
else:
|
|
return list(g.tasks)[0] if size == 1 else g
|
|
else:
|
|
return g.tasks[0] if size == 1 else g
|
|
|
|
|
|
class Signature(dict):
|
|
"""Class that wraps the arguments and execution options
|
|
for a single task invocation.
|
|
|
|
Used as the parts in a :class:`group` and other constructs,
|
|
or to pass tasks around as callbacks while being compatible
|
|
with serializers with a strict type subset.
|
|
|
|
:param task: Either a task class/instance, or the name of a task.
|
|
:keyword args: Positional arguments to apply.
|
|
:keyword kwargs: Keyword arguments to apply.
|
|
:keyword options: Additional options to :meth:`Task.apply_async`.
|
|
|
|
Note that if the first argument is a :class:`dict`, the other
|
|
arguments will be ignored and the values in the dict will be used
|
|
instead.
|
|
|
|
>>> s = signature('tasks.add', args=(2, 2))
|
|
>>> signature(s)
|
|
{'task': 'tasks.add', args=(2, 2), kwargs={}, options={}}
|
|
|
|
"""
|
|
TYPES = {}
|
|
_app = _type = None
|
|
|
|
@classmethod
|
|
def register_type(cls, subclass, name=None):
|
|
cls.TYPES[name or subclass.__name__] = subclass
|
|
return subclass
|
|
|
|
@classmethod
|
|
def from_dict(self, d, app=None):
|
|
typ = d.get('subtask_type')
|
|
if typ:
|
|
return self.TYPES[typ].from_dict(kwdict(d), app=app)
|
|
return Signature(d, app=app)
|
|
|
|
def __init__(self, task=None, args=None, kwargs=None, options=None,
|
|
type=None, subtask_type=None, immutable=False,
|
|
app=None, **ex):
|
|
self._app = app
|
|
init = dict.__init__
|
|
|
|
if isinstance(task, dict):
|
|
return init(self, task) # works like dict(d)
|
|
|
|
# Also supports using task class/instance instead of string name.
|
|
try:
|
|
task_name = task.name
|
|
except AttributeError:
|
|
task_name = task
|
|
else:
|
|
self._type = task
|
|
|
|
init(self,
|
|
task=task_name, args=tuple(args or ()),
|
|
kwargs=kwargs or {},
|
|
options=dict(options or {}, **ex),
|
|
subtask_type=subtask_type,
|
|
immutable=immutable)
|
|
|
|
def __call__(self, *partial_args, **partial_kwargs):
|
|
args, kwargs, _ = self._merge(partial_args, partial_kwargs, None)
|
|
return self.type(*args, **kwargs)
|
|
|
|
def delay(self, *partial_args, **partial_kwargs):
|
|
return self.apply_async(partial_args, partial_kwargs)
|
|
|
|
def apply(self, args=(), kwargs={}, **options):
|
|
"""Apply this task locally."""
|
|
# For callbacks: extra args are prepended to the stored args.
|
|
args, kwargs, options = self._merge(args, kwargs, options)
|
|
return self.type.apply(args, kwargs, **options)
|
|
|
|
def _merge(self, args=(), kwargs={}, options={}):
|
|
if self.immutable:
|
|
return (self.args, self.kwargs,
|
|
dict(self.options, **options) if options else self.options)
|
|
return (tuple(args) + tuple(self.args) if args else self.args,
|
|
dict(self.kwargs, **kwargs) if kwargs else self.kwargs,
|
|
dict(self.options, **options) if options else self.options)
|
|
|
|
def clone(self, args=(), kwargs={}, **opts):
|
|
# need to deepcopy options so origins links etc. is not modified.
|
|
if args or kwargs or opts:
|
|
args, kwargs, opts = self._merge(args, kwargs, opts)
|
|
else:
|
|
args, kwargs, opts = self.args, self.kwargs, self.options
|
|
s = Signature.from_dict({'task': self.task, 'args': tuple(args),
|
|
'kwargs': kwargs, 'options': deepcopy(opts),
|
|
'subtask_type': self.subtask_type,
|
|
'immutable': self.immutable}, app=self._app)
|
|
s._type = self._type
|
|
return s
|
|
partial = clone
|
|
|
|
def freeze(self, _id=None, group_id=None, chord=None):
|
|
opts = self.options
|
|
try:
|
|
tid = opts['task_id']
|
|
except KeyError:
|
|
tid = opts['task_id'] = _id or uuid()
|
|
if 'reply_to' not in opts:
|
|
opts['reply_to'] = self.app.oid
|
|
if group_id:
|
|
opts['group_id'] = group_id
|
|
if chord:
|
|
opts['chord'] = chord
|
|
return self.AsyncResult(tid)
|
|
_freeze = freeze
|
|
|
|
def replace(self, args=None, kwargs=None, options=None):
|
|
s = self.clone()
|
|
if args is not None:
|
|
s.args = args
|
|
if kwargs is not None:
|
|
s.kwargs = kwargs
|
|
if options is not None:
|
|
s.options = options
|
|
return s
|
|
|
|
def set(self, immutable=None, **options):
|
|
if immutable is not None:
|
|
self.set_immutable(immutable)
|
|
self.options.update(options)
|
|
return self
|
|
|
|
def set_immutable(self, immutable):
|
|
self.immutable = immutable
|
|
|
|
def apply_async(self, args=(), kwargs={}, **options):
|
|
try:
|
|
_apply = self._apply_async
|
|
except IndexError: # no tasks for chain, etc to find type
|
|
return
|
|
# For callbacks: extra args are prepended to the stored args.
|
|
if args or kwargs or options:
|
|
args, kwargs, options = self._merge(args, kwargs, options)
|
|
else:
|
|
args, kwargs, options = self.args, self.kwargs, self.options
|
|
return _apply(args, kwargs, **options)
|
|
|
|
def append_to_list_option(self, key, value):
|
|
items = self.options.setdefault(key, [])
|
|
if not isinstance(items, MutableSequence):
|
|
items = self.options[key] = [items]
|
|
if value not in items:
|
|
items.append(value)
|
|
return value
|
|
|
|
def link(self, callback):
|
|
return self.append_to_list_option('link', callback)
|
|
|
|
def link_error(self, errback):
|
|
return self.append_to_list_option('link_error', errback)
|
|
|
|
def flatten_links(self):
|
|
return list(_chain.from_iterable(_chain(
|
|
[[self]],
|
|
(link.flatten_links()
|
|
for link in maybe_list(self.options.get('link')) or [])
|
|
)))
|
|
|
|
def __or__(self, other):
|
|
if isinstance(other, group):
|
|
other = maybe_unroll_group(other)
|
|
if not isinstance(self, chain) and isinstance(other, chain):
|
|
return chain((self, ) + other.tasks, app=self._app)
|
|
elif isinstance(other, chain):
|
|
return chain(*self.tasks + other.tasks, app=self._app)
|
|
elif isinstance(other, Signature):
|
|
if isinstance(self, chain):
|
|
return chain(*self.tasks + (other, ), app=self._app)
|
|
return chain(self, other, app=self._app)
|
|
return NotImplemented
|
|
|
|
def __deepcopy__(self, memo):
|
|
memo[id(self)] = self
|
|
return dict(self)
|
|
|
|
def __invert__(self):
|
|
return self.apply_async().get()
|
|
|
|
def __reduce__(self):
|
|
# for serialization, the task type is lazily loaded,
|
|
# and not stored in the dict itself.
|
|
return subtask, (dict(self), )
|
|
|
|
def reprcall(self, *args, **kwargs):
|
|
args, kwargs, _ = self._merge(args, kwargs, {})
|
|
return reprcall(self['task'], args, kwargs)
|
|
|
|
def election(self):
|
|
type = self.type
|
|
app = type.app
|
|
tid = self.options.get('task_id') or uuid()
|
|
|
|
with app.producer_or_acquire(None) as P:
|
|
props = type.backend.on_task_call(P, tid)
|
|
app.control.election(tid, 'task', self.clone(task_id=tid, **props),
|
|
connection=P.connection)
|
|
return type.AsyncResult(tid)
|
|
|
|
def __repr__(self):
|
|
return self.reprcall()
|
|
|
|
@cached_property
|
|
def type(self):
|
|
return self._type or self.app.tasks[self['task']]
|
|
|
|
@cached_property
|
|
def app(self):
|
|
return self._app or current_app
|
|
|
|
@cached_property
|
|
def AsyncResult(self):
|
|
try:
|
|
return self.type.AsyncResult
|
|
except KeyError: # task not registered
|
|
return self.app.AsyncResult
|
|
|
|
@cached_property
|
|
def _apply_async(self):
|
|
try:
|
|
return self.type.apply_async
|
|
except KeyError:
|
|
return _partial(self.app.send_task, self['task'])
|
|
id = _getitem_property('options.task_id')
|
|
task = _getitem_property('task')
|
|
args = _getitem_property('args')
|
|
kwargs = _getitem_property('kwargs')
|
|
options = _getitem_property('options')
|
|
subtask_type = _getitem_property('subtask_type')
|
|
immutable = _getitem_property('immutable')
|
|
|
|
|
|
@Signature.register_type
|
|
class chain(Signature):
|
|
|
|
def __init__(self, *tasks, **options):
|
|
tasks = (regen(tasks[0]) if len(tasks) == 1 and is_list(tasks[0])
|
|
else tasks)
|
|
Signature.__init__(
|
|
self, 'celery.chain', (), {'tasks': tasks}, **options
|
|
)
|
|
self.tasks = tasks
|
|
self.subtask_type = 'chain'
|
|
|
|
def __call__(self, *args, **kwargs):
|
|
if self.tasks:
|
|
return self.apply_async(args, kwargs)
|
|
|
|
@classmethod
|
|
def from_dict(self, d, app=None):
|
|
tasks = d['kwargs']['tasks']
|
|
if d['args'] and tasks:
|
|
# partial args passed on to first task in chain (Issue #1057).
|
|
tasks[0]['args'] = tasks[0]._merge(d['args'])[0]
|
|
return chain(*d['kwargs']['tasks'], app=app, **kwdict(d['options']))
|
|
|
|
@property
|
|
def type(self):
|
|
try:
|
|
return self._type or self.tasks[0].type.app.tasks['celery.chain']
|
|
except KeyError:
|
|
return self.app.tasks['celery.chain']
|
|
|
|
def __repr__(self):
|
|
return ' | '.join(repr(t) for t in self.tasks)
|
|
|
|
|
|
class _basemap(Signature):
|
|
_task_name = None
|
|
_unpack_args = itemgetter('task', 'it')
|
|
|
|
def __init__(self, task, it, **options):
|
|
Signature.__init__(
|
|
self, self._task_name, (),
|
|
{'task': task, 'it': regen(it)}, immutable=True, **options
|
|
)
|
|
|
|
def apply_async(self, args=(), kwargs={}, **opts):
|
|
# need to evaluate generators
|
|
task, it = self._unpack_args(self.kwargs)
|
|
return self.type.apply_async(
|
|
(), {'task': task, 'it': list(it)}, **opts
|
|
)
|
|
|
|
@classmethod
|
|
def from_dict(cls, d, app=None):
|
|
return cls(*cls._unpack_args(d['kwargs']), app=app, **d['options'])
|
|
|
|
|
|
@Signature.register_type
|
|
class xmap(_basemap):
|
|
_task_name = 'celery.map'
|
|
|
|
def __repr__(self):
|
|
task, it = self._unpack_args(self.kwargs)
|
|
return '[{0}(x) for x in {1}]'.format(task.task,
|
|
truncate(repr(it), 100))
|
|
|
|
|
|
@Signature.register_type
|
|
class xstarmap(_basemap):
|
|
_task_name = 'celery.starmap'
|
|
|
|
def __repr__(self):
|
|
task, it = self._unpack_args(self.kwargs)
|
|
return '[{0}(*x) for x in {1}]'.format(task.task,
|
|
truncate(repr(it), 100))
|
|
|
|
|
|
@Signature.register_type
|
|
class chunks(Signature):
|
|
_unpack_args = itemgetter('task', 'it', 'n')
|
|
|
|
def __init__(self, task, it, n, **options):
|
|
Signature.__init__(
|
|
self, 'celery.chunks', (),
|
|
{'task': task, 'it': regen(it), 'n': n},
|
|
immutable=True, **options
|
|
)
|
|
|
|
@classmethod
|
|
def from_dict(self, d, app=None):
|
|
return chunks(*self._unpack_args(d['kwargs']), app=app, **d['options'])
|
|
|
|
def apply_async(self, args=(), kwargs={}, **opts):
|
|
return self.group().apply_async(args, kwargs, **opts)
|
|
|
|
def __call__(self, **options):
|
|
return self.group()(**options)
|
|
|
|
def group(self):
|
|
# need to evaluate generators
|
|
task, it, n = self._unpack_args(self.kwargs)
|
|
return group((xstarmap(task, part, app=self._app)
|
|
for part in _chunks(iter(it), n)),
|
|
app=self._app)
|
|
|
|
@classmethod
|
|
def apply_chunks(cls, task, it, n, app=None):
|
|
return cls(task, it, n, app=app)()
|
|
|
|
|
|
def _maybe_group(tasks):
|
|
if isinstance(tasks, group):
|
|
tasks = list(tasks.tasks)
|
|
elif isinstance(tasks, Signature):
|
|
tasks = [tasks]
|
|
else:
|
|
tasks = regen(tasks)
|
|
return tasks
|
|
|
|
|
|
def _maybe_clone(tasks, app):
|
|
return [s.clone() if isinstance(s, Signature) else signature(s, app=app)
|
|
for s in tasks]
|
|
|
|
|
|
@Signature.register_type
|
|
class group(Signature):
|
|
|
|
def __init__(self, *tasks, **options):
|
|
if len(tasks) == 1:
|
|
tasks = _maybe_group(tasks[0])
|
|
Signature.__init__(
|
|
self, 'celery.group', (), {'tasks': tasks}, **options
|
|
)
|
|
self.tasks, self.subtask_type = tasks, 'group'
|
|
|
|
@classmethod
|
|
def from_dict(self, d, app=None):
|
|
tasks = d['kwargs']['tasks']
|
|
if d['args'] and tasks:
|
|
# partial args passed on to all tasks in the group (Issue #1057).
|
|
for task in tasks:
|
|
task['args'] = task._merge(d['args'])[0]
|
|
return group(tasks, app=app, **kwdict(d['options']))
|
|
|
|
def apply_async(self, args=(), kwargs=None, add_to_parent=True, **options):
|
|
tasks = _maybe_clone(self.tasks, app=self._app)
|
|
if not tasks:
|
|
return self.freeze()
|
|
type = self.type
|
|
return type(*type.prepare(dict(self.options, **options), tasks, args),
|
|
add_to_parent=add_to_parent)
|
|
|
|
def set_immutable(self, immutable):
|
|
for task in self.tasks:
|
|
task.set_immutable(immutable)
|
|
|
|
def link(self, sig):
|
|
# Simply link to first task
|
|
sig = sig.clone().set(immutable=True)
|
|
return self.tasks[0].link(sig)
|
|
|
|
def link_error(self, sig):
|
|
sig = sig.clone().set(immutable=True)
|
|
return self.tasks[0].link_error(sig)
|
|
|
|
def apply(self, *args, **kwargs):
|
|
if not self.tasks:
|
|
return self.freeze() # empty group returns GroupResult
|
|
return Signature.apply(self, *args, **kwargs)
|
|
|
|
def __call__(self, *partial_args, **options):
|
|
return self.apply_async(partial_args, **options)
|
|
|
|
def freeze(self, _id=None, group_id=None, chord=None):
|
|
opts = self.options
|
|
try:
|
|
gid = opts['task_id']
|
|
except KeyError:
|
|
gid = opts['task_id'] = uuid()
|
|
if group_id:
|
|
opts['group_id'] = group_id
|
|
if chord:
|
|
opts['chord'] = group_id
|
|
new_tasks, results = [], []
|
|
for task in self.tasks:
|
|
task = maybe_signature(task, app=self._app).clone()
|
|
results.append(task.freeze(group_id=group_id, chord=chord))
|
|
new_tasks.append(task)
|
|
self.tasks = self.kwargs['tasks'] = new_tasks
|
|
return self.app.GroupResult(gid, results)
|
|
_freeze = freeze
|
|
|
|
def skew(self, start=1.0, stop=None, step=1.0):
|
|
it = fxrange(start, stop, step, repeatlast=True)
|
|
for task in self.tasks:
|
|
task.set(countdown=next(it))
|
|
return self
|
|
|
|
def __iter__(self):
|
|
return iter(self.tasks)
|
|
|
|
def __repr__(self):
|
|
return repr(self.tasks)
|
|
|
|
@property
|
|
def type(self):
|
|
if self._type:
|
|
return self._type
|
|
# taking the app from the first task in the list, there may be a
|
|
# better solution for this, e.g. to consolidate tasks with the same
|
|
# app and apply them in batches.
|
|
app = self._app if self._app else self.tasks[0].type.app
|
|
return app.tasks[self['task']]
|
|
|
|
|
|
@Signature.register_type
|
|
class chord(Signature):
|
|
|
|
def __init__(self, header, body=None, task='celery.chord',
|
|
args=(), kwargs={}, **options):
|
|
Signature.__init__(
|
|
self, task, args,
|
|
dict(kwargs, header=_maybe_group(header),
|
|
body=maybe_signature(body, app=self._app)), **options
|
|
)
|
|
self.subtask_type = 'chord'
|
|
|
|
def freeze(self, _id=None, group_id=None, chord=None):
|
|
return self.body.freeze(_id, group_id=group_id, chord=chord)
|
|
|
|
@classmethod
|
|
def from_dict(self, d, app=None):
|
|
args, d['kwargs'] = self._unpack_args(**kwdict(d['kwargs']))
|
|
return self(*args, app=app, **kwdict(d))
|
|
|
|
@staticmethod
|
|
def _unpack_args(header=None, body=None, **kwargs):
|
|
# Python signatures are better at extracting keys from dicts
|
|
# than manually popping things off.
|
|
return (header, body), kwargs
|
|
|
|
@property
|
|
def type(self):
|
|
if self._type:
|
|
return self._type
|
|
# we will be able to fix this mess in 3.2 when we no longer
|
|
# require an actual task implementation for chord/group
|
|
if self._app:
|
|
app = self._app
|
|
else:
|
|
try:
|
|
app = self.tasks[0].type.app
|
|
except IndexError:
|
|
app = self.body.type.app
|
|
return app.tasks['celery.chord']
|
|
|
|
def apply_async(self, args=(), kwargs={}, task_id=None,
|
|
producer=None, publisher=None, connection=None,
|
|
router=None, result_cls=None, **options):
|
|
body = kwargs.get('body') or self.kwargs['body']
|
|
kwargs = dict(self.kwargs, **kwargs)
|
|
body = body.clone(**options)
|
|
|
|
_chord = self.type
|
|
if _chord.app.conf.CELERY_ALWAYS_EAGER:
|
|
return self.apply((), kwargs, task_id=task_id, **options)
|
|
res = body.freeze(task_id)
|
|
parent = _chord(self.tasks, body, args, **options)
|
|
res.parent = parent
|
|
return res
|
|
|
|
def __call__(self, body=None, **options):
|
|
return self.apply_async((), {'body': body} if body else {}, **options)
|
|
|
|
def clone(self, *args, **kwargs):
|
|
s = Signature.clone(self, *args, **kwargs)
|
|
# need to make copy of body
|
|
try:
|
|
s.kwargs['body'] = s.kwargs['body'].clone()
|
|
except (AttributeError, KeyError):
|
|
pass
|
|
return s
|
|
|
|
def link(self, callback):
|
|
self.body.link(callback)
|
|
return callback
|
|
|
|
def link_error(self, errback):
|
|
self.body.link_error(errback)
|
|
return errback
|
|
|
|
def set_immutable(self, immutable):
|
|
# changes mutability of header only, not callback.
|
|
for task in self.tasks:
|
|
task.set_immutable(immutable)
|
|
|
|
def __repr__(self):
|
|
if self.body:
|
|
return self.body.reprcall(self.tasks)
|
|
return '<chord without body: {0.tasks!r}>'.format(self)
|
|
|
|
tasks = _getitem_property('kwargs.header')
|
|
body = _getitem_property('kwargs.body')
|
|
|
|
|
|
def signature(varies, *args, **kwargs):
|
|
if isinstance(varies, dict):
|
|
if isinstance(varies, Signature):
|
|
return varies.clone()
|
|
return Signature.from_dict(varies)
|
|
return Signature(varies, *args, **kwargs)
|
|
subtask = signature # XXX compat
|
|
|
|
|
|
def maybe_signature(d, app=None):
|
|
if d is not None:
|
|
if isinstance(d, dict):
|
|
if not isinstance(d, Signature):
|
|
return signature(d, app=app)
|
|
elif isinstance(d, list):
|
|
return [maybe_signature(s, app=app) for s in d]
|
|
if app is not None:
|
|
d._app = app
|
|
return d
|
|
|
|
maybe_subtask = maybe_signature # XXX compat
|