reset_value = task_join_will_block()

_set_task_join_will_block(False)

try:

yield

finally:

reset_value = task_join_will_block()

_set_task_join_will_block(True)

try:

yield

finally:

"""Base class for results."""

#: Parent result (if part of a chain)

"""Query task state.

app = None

#: Error raised for timeouts.

TimeoutError = TimeoutError

#: The task's UUID.

id = None

#: The task result backend to use.

backend = None

def __init__(self, id, backend=None,

task_name=None, # deprecated

app=None, parent=None):

if id is None:

raise ValueError(

f'AsyncResult requires valid id, not {type(id)}')

self.app = app_or_default(app or self.app)

self.id = id

self.backend = backend or self.app.backend

self.parent = parent

self.on_ready = promise(self._on_fulfilled, weak=True)

self._cache = None

self._ignored = False

@property

def ignored(self):

"""If True, task result retrieval is disabled."""

if hasattr(self, '_ignored'):

return self._ignored

return False

@ignored.setter

def ignored(self, value):

"""Enable/disable task result retrieval."""

self._ignored = value

def then(self, callback, on_error=None, weak=False):

self.backend.add_pending_result(self, weak=weak)

return self.on_ready.then(callback, on_error)

def _on_fulfilled(self, result):

self.backend.remove_pending_result(self)

return result

def as_tuple(self):

parent = self.parent

return (self.id, parent and parent.as_tuple()), None

def as_list(self):

"""Return as a list of task IDs."""

results = []

parent = self.parent

results.append(self.id)

if parent is not None:

results.extend(parent.as_list())

return results

def forget(self):

"""Forget the result of this task and its parents."""

self._cache = None

if self.parent:

self.parent.forget()

self.backend.remove_pending_result(self)

self.backend.forget(self.id)

def revoke(self, connection=None, terminate=False, signal=None,

wait=False, timeout=None):

"""Send revoke signal to all workers.

self.app.control.revoke(self.id, connection=connection,

terminate=terminate, signal=signal,

reply=wait, timeout=timeout)

def revoke_by_stamped_headers(self, headers, connection=None, terminate=False, signal=None,

wait=False, timeout=None):

"""Send revoke signal to all workers only for tasks with matching headers values.

self.app.control.revoke_by_stamped_headers(headers, connection=connection,

terminate=terminate, signal=signal,

reply=wait, timeout=timeout)

def get(self, timeout=None, propagate=True, interval=0.5,

no_ack=True, follow_parents=True, callback=None, on_message=None,

on_interval=None, disable_sync_subtasks=True,

EXCEPTION_STATES=states.EXCEPTION_STATES,

PROPAGATE_STATES=states.PROPAGATE_STATES):

"""Wait until task is ready, and return its result.

if self.ignored:

return

if disable_sync_subtasks:

assert_will_not_block()

_on_interval = promise()

if follow_parents and propagate and self.parent:

_on_interval = promise(self._maybe_reraise_parent_error, weak=True)

self._maybe_reraise_parent_error()

if on_interval:

_on_interval.then(on_interval)

if self._cache:

if propagate:

self.maybe_throw(callback=callback)

return self.result

self.backend.add_pending_result(self)

return self.backend.wait_for_pending(

self, timeout=timeout,

interval=interval,

on_interval=_on_interval,

no_ack=no_ack,

propagate=propagate,

callback=callback,

on_message=on_message,

)

wait = get # deprecated alias to :meth:`get`.

def _maybe_reraise_parent_error(self):

for node in reversed(list(self._parents())):

node.maybe_throw()

def _parents(self):

node = self.parent

while node:

yield node

node = node.parent

def collect(self, intermediate=False, **kwargs):

"""Collect results as they return.

for _, R in self.iterdeps(intermediate=intermediate):

yield R, R.get(**kwargs)

def get_leaf(self):

value = None

for _, R in self.iterdeps():

value = R.get()

return value

def iterdeps(self, intermediate=False):

stack = deque([(None, self)])

is_incomplete_stream = not intermediate

while stack:

parent, node = stack.popleft()

yield parent, node

if node.ready():

stack.extend((node, child) for child in node.children or [])

else:

if is_incomplete_stream:

raise IncompleteStream()

def exists(self):

"""Return :const:`True` if a result exists in the backend for this task.

return self.backend.task_result_exists(self.id)

def ready(self):

"""Return :const:`True` if the task has executed.

return self.state in self.backend.READY_STATES

def successful(self):

"""Return :const:`True` if the task executed successfully."""

return self.state == states.SUCCESS

def failed(self):

"""Return :const:`True` if the task failed."""

return self.state == states.FAILURE

def throw(self, *args, **kwargs):

self.on_ready.throw(*args, **kwargs)

def maybe_throw(self, propagate=True, callback=None):

cache = self._get_task_meta() if self._cache is None else self._cache

state, value, tb = (

cache['status'], cache['result'], cache.get('traceback'))

if state in states.PROPAGATE_STATES and propagate:

self.throw(value, self._to_remote_traceback(tb))

if callback is not None:

callback(self.id, value)

return value

maybe_reraise = maybe_throw # XXX compat alias

def _to_remote_traceback(self, tb):

if tb and tblib is not None and self.app.conf.task_remote_tracebacks:

return tblib.Traceback.from_string(tb).as_traceback()

def build_graph(self, intermediate=False, formatter=None):

graph = DependencyGraph(

formatter=formatter or GraphFormatter(root=self.id, shape='oval'),

)

for parent, node in self.iterdeps(intermediate=intermediate):

graph.add_arc(node)

if parent:

graph.add_edge(parent, node)

return graph

__class_getitem__ = classmethod(types.GenericAlias)

def __str__(self):

"""`str(self) -> self.id`."""

return str(self.id)

def __hash__(self):

"""`hash(self) -> hash(self.id)`."""

return hash(self.id)

def __repr__(self):

return f'<{type(self).__name__}: {self.id}>'

def __eq__(self, other):

if isinstance(other, AsyncResult):

return other.id == self.id

elif isinstance(other, str):

return other == self.id

return NotImplemented

def __copy__(self):

return self.__class__(

self.id, self.backend, None, self.app, self.parent,

)

def __reduce__(self):

return self.__class__, self.__reduce_args__()

def __reduce_args__(self):

return self.id, self.backend, None, None, self.parent

def __del__(self):

"""Cancel pending operations when the instance is destroyed."""

if self.backend is not None:

self.backend.remove_pending_result(self)

@cached_property

def graph(self):

return self.build_graph()

@property

def supports_native_join(self):

return self.backend.supports_native_join

@property

def children(self):

return self._get_task_meta().get('children')

def _maybe_set_cache(self, meta):

if meta:

state = meta['status']

if state in states.READY_STATES:

d = self._set_cache(self.backend.meta_from_decoded(meta))

self.on_ready(self)

return d

return meta

def _get_task_meta(self):

if self._cache is None:

return self._maybe_set_cache(self.backend.get_task_meta(self.id))

return self._cache

def _iter_meta(self, **kwargs):

return iter([self._get_task_meta()])

def _set_cache(self, d):

children = d.get('children')

if children:

d['children'] = [

result_from_tuple(child, self.app) for child in children

]

self._cache = d

return d

@property

def result(self):

"""Task return value.

return self._get_task_meta()['result']

info = result

@property

def traceback(self):

"""Get the traceback of a failed task."""

return self._get_task_meta().get('traceback')

@property

def state(self):

"""The tasks current state.

return self._get_task_meta()['status']

status = state # XXX compat

@property

def task_id(self):

"""Compat. alias to :attr:`id`."""

return self.id

@task_id.setter

def task_id(self, id):

self.id = id

@property

def name(self):

return self._get_task_meta().get('name')

@property

def args(self):

return self._get_task_meta().get('args')

@property

def kwargs(self):

return self._get_task_meta().get('kwargs')

@property

def worker(self):

return self._get_task_meta().get('worker')

@property

def date_done(self):

"""UTC date and time."""

date_done = self._get_task_meta().get('date_done')

if date_done and not isinstance(date_done, datetime.datetime):

return isoparse(date_done)

return date_done

@property

def retries(self):

return self._get_task_meta().get('retries')

@property

def queue(self):

"""A collection of results.

_app = None

#: List of results in in the set.

results = None

def __init__(self, results, app=None, ready_barrier=None, **kwargs):

self._app = app

self.results = results

self.on_ready = promise(args=(proxy(self),))

self._on_full = ready_barrier or barrier(results)

if self._on_full:

self._on_full.then(promise(self._on_ready, weak=True))

def add(self, result):

"""Add :class:`AsyncResult` as a new member of the set.

if result not in self.results:

self.results.append(result)

if self._on_full:

self._on_full.add(result)

def _on_ready(self):

if self.backend.is_async:

self.on_ready()

def remove(self, result):

"""Remove result from the set; it must be a member.

if isinstance(result, str):

result = self.app.AsyncResult(result)

try:

self.results.remove(result)

except ValueError:

raise KeyError(result)

def discard(self, result):

"""Remove result from the set if it is a member.

try:

self.remove(result)

except KeyError:

pass

def update(self, results):

"""Extend from iterable of results."""

self.results.extend(r for r in results if r not in self.results)

def clear(self):

"""Remove all results from this set."""

self.results[:] = [] # don't create new list.

def successful(self):

"""Return true if all tasks successful.

return all(result.successful() for result in self.results)

def failed(self):

"""Return true if any of the tasks failed.

return any(result.failed() for result in self.results)

def maybe_throw(self, callback=None, propagate=True):

for result in self.results:

result.maybe_throw(callback=callback, propagate=propagate)

maybe_reraise = maybe_throw # XXX compat alias.

def waiting(self):

"""Return true if any of the tasks are incomplete.

return any(not result.ready() for result in self.results)

def ready(self):

"""Did all of the tasks complete? (either by success of failure).

return all(result.ready() for result in self.results)

def completed_count(self):

"""Task completion count.

return sum(int(result.successful()) for result in self.results)

def forget(self):

"""Forget about (and possible remove the result of) all the tasks."""

for result in self.results:

result.forget()

def revoke(self, connection=None, terminate=False, signal=None,

wait=False, timeout=None):

"""Send revoke signal to all workers for all tasks in the set.

self.app.control.revoke([r.id for r in self.results],

connection=connection, timeout=timeout,

terminate=terminate, signal=signal, reply=wait)

def __iter__(self):

return iter(self.results)

def __getitem__(self, index):

"""`res[i] -> res.results[i]`."""

return self.results[index]

def get(self, timeout=None, propagate=True, interval=0.5,

callback=None, no_ack=True, on_message=None,

disable_sync_subtasks=True, on_interval=None):

"""See :meth:`join`.

return (self.join_native if self.supports_native_join else self.join)(

timeout=timeout, propagate=propagate,

interval=interval, callback=callback, no_ack=no_ack,

on_message=on_message, disable_sync_subtasks=disable_sync_subtasks,

on_interval=on_interval,

)

def join(self, timeout=None, propagate=True, interval=0.5,

callback=None, no_ack=True, on_message=None,

disable_sync_subtasks=True, on_interval=None):

"""Gather the results of all tasks as a list in order.

if disable_sync_subtasks:

assert_will_not_block()

time_start = time.monotonic()

remaining = None

if on_message is not None:

raise ImproperlyConfigured(

'Backend does not support on_message callback')

results = []

for result in self.results:

remaining = None

if timeout:

remaining = timeout - (time.monotonic() - time_start)

if remaining <= 0.0:

raise TimeoutError('join operation timed out')

value = result.get(

timeout=remaining, propagate=propagate,

interval=interval, no_ack=no_ack, on_interval=on_interval,

disable_sync_subtasks=disable_sync_subtasks,

)

if callback:

callback(result.id, value)

else:

results.append(value)

return results

def then(self, callback, on_error=None, weak=False):

return self.on_ready.then(callback, on_error)

def iter_native(self, timeout=None, interval=0.5, no_ack=True,

on_message=None, on_interval=None):

"""Backend optimized version of :meth:`iterate`.

return self.backend.iter_native(

self,

timeout=timeout, interval=interval, no_ack=no_ack,

on_message=on_message, on_interval=on_interval,

)

def join_native(self, timeout=None, propagate=True,

interval=0.5, callback=None, no_ack=True,

on_message=None, on_interval=None,

disable_sync_subtasks=True):

"""Backend optimized version of :meth:`join`.

if disable_sync_subtasks:

assert_will_not_block()

order_index = None if callback else {

result.id: i for i, result in enumerate(self.results)

}

acc = None if callback else [None for _ in range(len(self))]

for task_id, meta in self.iter_native(timeout, interval, no_ack,

on_message, on_interval):

if isinstance(meta, list):

value = []

for children_result in meta:

value.append(children_result.get())

else:

value = meta['result']

if propagate and meta['status'] in states.PROPAGATE_STATES:

raise value

if callback:

callback(task_id, value)

else:

acc[order_index[task_id]] = value

return acc

def _iter_meta(self, **kwargs):

return (meta for _, meta in self.backend.get_many(

{r.id for r in self.results}, max_iterations=1, **kwargs

))

def _failed_join_report(self):

return (res for res in self.results

if res.backend.is_cached(res.id) and

res.state in states.PROPAGATE_STATES)

def __len__(self):

return len(self.results)

def __eq__(self, other):

if isinstance(other, ResultSet):

return other.results == self.results

return NotImplemented

def __repr__(self):

return f'<{type(self).__name__}: [{", ".join(r.id for r in self.results)}]>'

@property

def supports_native_join(self):

try:

return self.results[0].supports_native_join

except IndexError:

pass

@property

def app(self):

if self._app is None:

self._app = (self.results[0].app if self.results else

current_app._get_current_object())

return self._app

@app.setter

def app(self, app):

self._app = app

@property

def backend(self):

"""Like :class:`ResultSet`, but with an associated id.

#: The UUID of the group.

id = None

#: List/iterator of results in the group

results = None

def __init__(self, id=None, results=None, parent=None, **kwargs):

self.id = id

self.parent = parent

super().__init__(results, **kwargs)

def _on_ready(self):

self.backend.remove_pending_result(self)

super()._on_ready()

def save(self, backend=None):

"""Save group-result for later retrieval using :meth:`restore`.

return (backend or self.app.backend).save_group(self.id, self)

def delete(self, backend=None):

"""Remove this result if it was previously saved."""

(backend or self.app.backend).delete_group(self.id)

def __reduce__(self):

return self.__class__, self.__reduce_args__()

def __reduce_args__(self):

return self.id, self.results

def __bool__(self):

return bool(self.id or self.results)

__nonzero__ = __bool__ # Included for Py2 backwards compatibility

def __eq__(self, other):

if isinstance(other, GroupResult):

return (

other.id == self.id and

other.results == self.results and

other.parent == self.parent

)

elif isinstance(other, str):

return other == self.id

return NotImplemented

def __repr__(self):

return f'<{type(self).__name__}: {self.id} [{", ".join(r.id for r in self.results)}]>'

def __str__(self):

"""`str(self) -> self.id`."""

return str(self.id)

def __hash__(self):