Merge pull request #96 from Koed00/dev

Adds task chains

django_q/cluster.py

@@ -328,7 +328,7 @@ def monitor(result_queue, broker=None):
         if task.get('cached', False):
             save_cached(task, broker)
         else:
-            save_task(task)
+            save_task(task, broker)
         # log the result
         if task['success']:
             logger.info(_("Processed [{}]").format(task['name']))
@@ -386,13 +386,16 @@ def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT):
     logger.info(_('{} stopped doing work').format(name))
 
 
-def save_task(task):
+def save_task(task, broker):
     """
     Saves the task package to Django or the cache
     """
     # SAVE LIMIT < 0 : Don't save success
     if not task.get('save', Conf.SAVE_LIMIT > 0) and task['success']:
         return
+    # async next in a chain
+    if task.get('chain', None):
+        tasks.async_chain(task['chain'], group=task['group'], cached=task['cached'], sync=task['sync'], broker=broker)
     # SAVE LIMIT > 0: Prune database, SAVE_LIMIT 0: No pruning
     db.close_old_connections()
     try:
@@ -419,15 +422,15 @@ def save_cached(task, broker):
     if timeout is True:
         timeout = None
     try:
-        group = task.get('group', False)
+        group = task.get('group', None)
         iter_count = task.get('iter_count', 0)
         # if it's a group append to the group list
         if group:
             task_key = '{}:{}:{}'.format(broker.list_key, group, task['id'])
             group_key = '{}:{}:keys'.format(broker.list_key, group)
             group_list = broker.cache.get(group_key) or []
-            # if it's an inter group, check if we are ready
-            if iter_count and len(group_list) == iter_count-1:
+            # if it's an iter group, check if we are ready
+            if iter_count and len(group_list) == iter_count - 1:
                 group_args = '{}:{}:args'.format(broker.list_key, group)
                 # collate the results into a Task result
                 results = [signing.SignedPackage.loads(broker.cache.get(k))['result'] for k in group_list]
@@ -441,13 +444,16 @@ def save_cached(task, broker):
                     task['cached'] = task.pop('iter_cached', None)
                     save_cached(task, broker=broker)
                 else:
-                    save_task(task)
+                    save_task(task, broker)
                 broker.cache.delete_many(group_list)
                 broker.cache.delete_many([group_key, group_args])
                 return
             # save the group list
             group_list.append(task_key)
             broker.cache.set(group_key, group_list)
+            # async next in a chain
+            if task.get('chain', None):
+                tasks.async_chain(task['chain'], group=group, cached=task['cached'], sync=task['sync'], broker=broker)
         # save the task
         broker.cache.set(task_key,
                          signing.SignedPackage.dumps(task),

django_q/models.py

@@ -79,7 +79,7 @@ class Task(models.Model):
         return (self.stopped - self.started).total_seconds()
 
     def __unicode__(self):
-        return self.name
+        return u'{}'.format(self.name or self.id)
 
     class Meta:
         app_label = 'django_q'

django_q/tasks.py

@@ -18,15 +18,19 @@ def async(func, *args, **kwargs):
     """Queue a task for the cluster."""
     # get options from q_options dict or direct from kwargs
     options = kwargs.pop('q_options', kwargs)
-    hook = options.pop('hook', None)
     broker = options.pop('broker', get_broker())
-    sync = options.pop('sync', False)
-    group = options.pop('group', None)
-    save = options.pop('save', None)
-    cached = options.pop('cached', Conf.CACHED)
-    iter_count = options.pop('iter_count', None)
-    iter_cached = options.pop('iter_cached', None)
-    # get an id
+    # pop optionals
+    opts = {'hook': None,
+            'group': None,
+            'save': None,
+            'sync': None,
+            'cached': Conf.CACHED,
+            'iter_count': None,
+            'iter_cached': None,
+            'chain': None}
+    for key in opts:
+        opts[key] = options.pop(key, opts[key])
+        # get an id
     tag = uuid()
     # build the task package
     task = {'id': tag[1], 'name': tag[0],
@@ -34,22 +38,13 @@ def async(func, *args, **kwargs):
             'args': args,
             'kwargs': kwargs,
             'started': timezone.now()}
-    # add optionals
-    if hook:
-        task['hook'] = hook
-    if group:
-        task['group'] = group
-    if save is not None:
-        task['save'] = save
-    if cached:
-        task['cached'] = cached
-    if iter_count:
-        task['iter_count'] = iter_count
-    if iter_cached:
-        task['iter_cached'] = iter_cached
+    # push optionals
+    for key in opts:
+        if opts[key] is not None:
+            task[key] = opts[key]
     # sign it
     pack = signing.SignedPackage.dumps(task)
-    if sync or Conf.SYNC:
+    if task.get('sync', False) or Conf.SYNC:
         return _sync(pack)
     # push it
     broker.enqueue(pack)
@@ -371,7 +366,7 @@ def delete_cached(task_id, broker=None):
 def queue_size(broker=None):
     """
     Returns the current queue size.
-    Note that this doesn't count any tasks curren    key = 'django_q:{}:results'.format(broker.list_key)tly being processed by workers.
+    Note that this doesn't count any tasks currently being processed by workers.
 
     :param broker: optional broker
     :return: current queue size
@@ -383,6 +378,9 @@ def queue_size(broker=None):
 
 
 def async_iter(func, args_iter, **kwargs):
+    """
+    async a function with iterable arguments
+    """
     iter_count = len(args_iter)
     iter_group = uuid()[1]
     # clean up the kwargs
@@ -404,6 +402,164 @@ def async_iter(func, args_iter, **kwargs):
     return iter_group
 
 
+def async_chain(chain, group=None, cached=Conf.CACHED, sync=Conf.SYNC, broker=None):
+    """
+    async a chain of tasks
+    the chain must be in the format [(func,(args),{kwargs}),(func,(args),{kwargs})]
+    """
+    if not group:
+        group = uuid()[1]
+    args = ()
+    kwargs = {}
+    task = chain.pop(0)
+    if type(task) is not tuple:
+        task = (task,)
+    if len(task) > 1:
+        args = task[1]
+    if len(task) > 2:
+        kwargs = task[2]
+    kwargs['chain'] = chain
+    kwargs['group'] = group
+    kwargs['cached'] = cached
+    kwargs['sync'] = sync
+    kwargs['broker'] = broker or get_broker()
+    async(task[0], *args, **kwargs)
+    return group
+
+
+class Iter(object):
+    """
+    An async task with iterable arguments
+    """
+
+    def __init__(self, func=None, args=None, kwargs=None, cached=Conf.CACHED, sync=Conf.SYNC, broker=None):
+        self.func = func
+        self.args = args or []
+        self.kwargs = kwargs or {}
+        self.id = ''
+        self.broker = broker or get_broker()
+        self.cached = cached
+        self.sync = sync
+        self.started = False
+
+    def append(self, *args):
+        """
+        add arguments to the set
+        """
+        self.args.append(args)
+        if self.started:
+            self.started = False
+        return self.length()
+
+    def run(self):
+        """
+        Start queueing the tasks to the worker cluster
+        :return: the task id
+        """
+        self.kwargs['cached'] = self.cached
+        self.kwargs['sync'] = self.sync
+        self.kwargs['broker'] = self.broker
+        self.id = async_iter(self.func, self.args, **self.kwargs)
+        self.started = True
+        return self.id
+
+    def result(self, wait=0):
+        """
+        return the full list of results.
+        :param int wait: how many milliseconds to wait for a result
+        :return: an unsorted list of results
+        """
+        if self.started:
+            return result(self.id, wait=wait, cached=self.cached)
+
+    def fetch(self,  wait=0):
+        """
+        get the task result objects.
+        :param int wait: how many milliseconds to wait for a result
+        :return: an unsorted list of task objects
+        """
+        if self.started:
+            return fetch(self.id, wait=wait, cached=self.cached)
+
+    def length(self):
+        """
+        get the length of the arguments list
+        :return int: length of the argument list
+        """
+        return len(self.args)
+
+
+class Chain(object):
+    """
+    A sequential chain of tasks
+    """
+
+    def __init__(self, chain=None, group=None, cached=Conf.CACHED, sync=Conf.SYNC):
+        self.chain = chain or []
+        self.group = group or ''
+        self.broker = get_broker()
+        self.cached = cached
+        self.sync = sync
+        self.started = False
+
+    def append(self, func, *args, **kwargs):
+        """
+        add a task to the chain
+        takes the same parameters as async()
+        """
+        self.chain.append((func, args, kwargs))
+        # remove existing results
+        if self.started:
+            delete_group(self.group)
+            self.started = False
+        return self.length()
+
+    def run(self):
+        """
+        Start queueing the chain to the worker cluster
+        :return: the chain's group id
+        """
+        self.group = async_chain(chain=self.chain[:], group=self.group, cached=self.cached, sync=self.sync,
+                                 broker=self.broker)
+        self.started = True
+        return self.group
+
+    def result(self, wait=0):
+        """
+        return the full list of results from the chain when it finishes. blocks until timeout.
+        :param int wait: how many milliseconds to wait for a result
+        :return: an unsorted list of results
+        """
+        if self.started:
+            return result_group(self.group, wait=wait, count=self.length(), cached=self.cached)
+
+    def fetch(self, failures=True, wait=0):
+        """
+        get the task result objects from the chain when it finishes. blocks until timeout.
+        :param failures: include failed tasks
+        :param int wait: how many milliseconds to wait for a result
+        :return: an unsorted list of task objects
+        """
+        if self.started:
+            return fetch_group(self.group, failures=failures, wait=wait, count=self.length(), cached=self.cached)
+
+    def current(self):
+        """
+        get the index of the currently executing chain element
+        :return int: current chain index
+        """
+        if not self.started:
+            return None
+        return count_group(self.group, cached=self.cached)
+
+    def length(self):
+        """
+        get the length of the chain
+        :return int: length of the chain
+        """
+        return len(self.chain)
+
+
 def _sync(pack):
     """Simulate a package travelling through the cluster."""
     task_queue = Queue()

django_q/tests/tasks.py

@@ -38,5 +38,9 @@ def get_user_id(user):
     return user.id
 
 
+def hello():
+    return 'hello'
+
+
 def result(obj):
     print('RESULT HOOK {} : {}'.format(obj.name, obj.result))

django_q/tests/test_cached.py

@@ -1,11 +1,11 @@
 from multiprocessing import Event, Queue, Value
 
 import pytest
-from django_q.cluster import pusher, worker, monitor
 
+from django_q.cluster import pusher, worker, monitor
 from django_q.conf import Conf
 from django_q.tasks import async, result, fetch, count_group, result_group, fetch_group, delete_group, delete_cached, \
-    async_iter
+    async_iter, Chain, async_chain, Iter
 from django_q.brokers import get_broker
 
 
@@ -96,9 +96,52 @@ def test_iter(broker):
     result_t = result(t)
     assert result_t is not None
     task_t = fetch(t)
-    assert task_t. __unicode__ is not None
     assert task_t.result == result_t
     assert result(t2) is not None
     assert result(t3) is not None
     assert result(t4)[0] == 1
-    # test cached iter result
+    # test iter class
+    i = Iter('math.copysign', sync=True, cached=True)
+    i.append(1, -1)
+    i.append(2, -1)
+    i.append(3, -4)
+    i.append(5, 6)
+    assert i.started is False
+    assert i.length() == 4
+    assert i.run() is not None
+    assert len(i.result()) == 4
+    assert len(i.fetch().result) == 4
+    i.append(1, -7)
+    assert i.result() is None
+    i.run()
+    assert len(i.result()) == 5
+
+
+@pytest.mark.django_db
+def test_chain(broker):
+    broker.purge_queue()
+    broker.cache.clear()
+    task_chain = Chain(sync=True)
+    task_chain.append('math.floor', 1)
+    task_chain.append('math.copysign', 1, -1)
+    task_chain.append('math.floor', 2)
+    assert task_chain.length() == 3
+    assert task_chain.current() is None
+    task_chain.run()
+    r = task_chain.result(wait=1000)
+    assert task_chain.current() == task_chain.length()
+    assert len(r) == task_chain.length()
+    t = task_chain.fetch()
+    assert len(t) == task_chain.length()
+    task_chain.cached = True
+    task_chain.append('math.floor', 3)
+    assert task_chain.length() == 4
+    task_chain.run()
+    r = task_chain.result(wait=1000)
+    assert task_chain.current() == task_chain.length()
+    assert len(r) == task_chain.length()
+    t = task_chain.fetch()
+    assert len(t) == task_chain.length()
+    # test single
+    rid = async_chain(['django_q.tests.tasks.hello', 'django_q.tests.tasks.hello'], sync=True, cached=True)
+    assert result_group(rid, cached=True) == ['hello', 'hello']

docs/architecture.rst

@@ -0,0 +1,91 @@
+Architecture
+------------
+
+.. image:: _static/cluster.png
+   :alt: Django Q schema
+
+
+Signed Tasks
+""""""""""""
+
+Tasks are first pickled and then signed using Django's own :mod:`django.core.signing` module using the ``SECRET_KEY`` and cluster name as salt, before being sent to a message broker. This ensures that task
+packages on the broker can only be executed and read by clusters
+and django servers who share the same secret key and cluster name.
+If a package fails to unpack, it will be marked failed with the broker and discarded.
+Optionally the packages can be compressed before transport.
+
+Broker
+""""""
+
+The broker collects task packages from the django instances and queues them for pick up by a cluster.
+If the broker supports message receipts, it will keep a copy of the tasks around until a cluster acknowledges the processing of the task.
+Otherwise it is put back in the queue after a timeout period. This ensure at-least-once delivery.
+Note that even if the task errors when processed by the cluster, this is considered a successful delivery.
+Most failed deliveries will be the result of a worker or the cluster crashing before the task was saved.
+
+Pusher
+""""""
+
+The pusher process continuously checks the broker for new task
+packages. It checks the signing and unpacks the task to the internal Task Queue.
+The amount of tasks in the Task Queue can be configured to control memory usage and minimize data loss in case of a failure.
+
+Worker
+""""""
+
+A worker process pulls a task of the Task Queue and it sets a shared countdown timer with :ref:`sentinel` indicating it is about to start work.
+The worker then tries to execute the task and afterwards the timer is reset and any results (including errors) are saved to the package.
+Irrespective of the failure or success of any of these steps, the package is then pushed onto the Result Queue.
+
+Monitor
+"""""""
+
+The result monitor checks the Result Queue for processed packages and
+saves both failed and successful packages to the Django database or cache backend.
+If the broker supports it, a delivery receipt is sent.
+In case the task was part of a chain, the next task is queued.
+
+.. _sentinel:
+
+Sentinel
+""""""""
+
+The sentinel spawns all process and then checks the health of all
+workers, including the pusher and the monitor. This includes checking timers on each worker for timeouts.
+In case of a sudden death or timeout, it will reincarnate the failing processes. When a stop signal is received, the sentinel will halt the
+pusher and instruct the workers and monitor to finish the remaining items. See :ref:`stop_procedure`
+
+Timeouts
+""""""""
+Before each task execution the worker sets a countdown timer on the sentinel and resets it again after execution.
+Meanwhile the sentinel checks if the timers don't reach zero, in which case it will terminate the worker and reincarnate a new one.
+
+Scheduler
+"""""""""
+Twice a minute the scheduler checks for any scheduled tasks that should be starting.
+
+- Creates a task from the schedule
+- Subtracts 1 from :attr:`django_q.Schedule.repeats`
+- Sets the next run time if there are repeats left or if it has a negative value.
+
+.. _stop_procedure:
+
+Stop procedure
+""""""""""""""
+
+When a stop signal is received, the sentinel exits the guard loop and instructs the pusher to stop pushing.
+Once this is confirmed, the sentinel pushes poison pills onto the task queue and will wait for all the workers to exit.
+This ensures that the task queue is emptied before the workers exit.
+Afterwards the sentinel waits for the monitor to empty the result queue and the stop procedure is complete.
+
+- Send stop event to pusher
+- Wait for pusher to exit
+- Put poison pills in the Task Queue
+- Wait for all the workers to clear the queue and stop
+- Put a poison pill on the Result Queue
+- Wait for monitor to process remaining results and exit
+- Signal that we have stopped
+
+.. warning::
+    If you force the cluster to terminate before the stop procedure has completed, you can lose tasks or results still being held in memory.
+    You can manage the amount of tasks in a clusters memory by setting the :ref:`queue_limit`.

docs/cluster.rst

@@ -82,87 +82,6 @@ An example :file:`circus.ini` ::
 Note that we only start one process. It is not a good idea to run multiple instances of the cluster in the same environment since this does nothing to increase performance and in all likelihood will diminish it.
 Control your cluster using the ``workers``, ``recycle`` and ``timeout`` settings in your :doc:`configure`
 
-Architecture
-------------
-
-.. image:: _static/cluster.png
-   :alt: Django Q schema
-
-
-Signed Tasks
-""""""""""""
-
-Tasks are first pickled and then signed using Django's own :mod:`django.core.signing` module using the ``SECRET_KEY`` and cluster name as salt, before being sent to a message broker. This ensures that task
-packages on the broker can only be executed and read by clusters
-and django servers who share the same secret key and cluster name.
-If a package fails to unpack, it will be marked failed with the broker and discarded.
-Optionally the packages can be compressed before transport.
-
-Pusher
-""""""
-
-The pusher process continuously checks the broker for new task
-packages. It checks the signing and unpacks the task to the Task Queue.
-
-Worker
-""""""
-
-A worker process pulls a task of the Task Queue and it sets a shared countdown timer with :ref:`sentinel` indicating it is about to start work.
-The worker then tries to execute the task and afterwards the timer is reset and any results (including errors) are saved to the package.
-Irrespective of the failure or success of any of these steps, the package is then pushed onto the Result Queue.
-
-Monitor
-"""""""
-
-The result monitor checks the Result Queue for processed packages and
-saves both failed and successful packages to the Django database.
-If the broker supports it, a delivery receipt is sent.
-
-.. _sentinel:
-
-Sentinel
-""""""""
-
-The sentinel spawns all process and then checks the health of all
-workers, including the pusher and the monitor. This includes checking timers on each worker for timeouts.
-In case of a sudden death or timeout, it will reincarnate the failing processes. When a stop signal is received, the sentinel will halt the
-pusher and instruct the workers and monitor to finish the remaining items. See :ref:`stop_procedure`
-
-Timeouts
-""""""""
-Before each task execution the worker sets a countdown timer on the sentinel and resets it again after execution.
-Meanwhile the sentinel checks if the timers don't reach zero, in which case it will terminate the worker and reincarnate a new one.
-
-Scheduler
-"""""""""
-Twice a minute the scheduler checks for any scheduled tasks that should be starting.
-
-- Creates a task from the schedule
-- Subtracts 1 from :attr:`django_q.Schedule.repeats`
-- Sets the next run time if there are repeats left or if it has a negative value.
-
-.. _stop_procedure:
-
-Stop procedure
-""""""""""""""
-
-When a stop signal is received, the sentinel exits the guard loop and instructs the pusher to stop pushing.
-Once this is confirmed, the sentinel pushes poison pills onto the task queue and will wait for all the workers to exit.
-This ensures that the task queue is emptied before the workers exit.
-Afterwards the sentinel waits for the monitor to empty the result queue and the stop procedure is complete.
-
-- Send stop event to pusher
-- Wait for pusher to exit
-- Put poison pills in the Task Queue
-- Wait for all the workers to clear the queue and stop
-- Put a poison pill on the Result Queue
-- Wait for monitor to process remaining results and exit
-- Signal that we have stopped
-
-.. warning::
-    If you force the cluster to terminate before the stop procedure has completed, you can lose tasks or results still being held in memory.
-    You can manage the amount of tasks in a clusters memory by setting the :ref:`queue_limit`.
-
 Reference
 ---------
 

docs/index.rst

@@ -11,7 +11,7 @@ Django Q is a native Django task queue, scheduler and worker application using P
 Features
 --------
 
--  Multiprocessing worker pool
+-  Multiprocessing worker poole7u-CG
 -  Asynchronous tasks
 -  Scheduled and repeated tasks
 -  Encrypted and compressed packages
@@ -39,6 +39,7 @@ Contents:
     Cluster <cluster>
     Monitor <monitor>
     Admin <admin>
+    Architecture <architecture>
     Examples <examples>
 
 * :ref:`genindex`

docs/tasks.rst

@@ -91,8 +91,8 @@ Please not that this will override any other option keywords.
 
 
 
-Async Iterable
---------------
+Iterable
+--------
 If you have an iterable object with arguments for a function, you can use :func:`async_iter` to async them with a single command::
 
     # Async Iterable example
@@ -109,6 +109,30 @@ If you have an iterable object with arguments for a function, you can use :func:
 
 This will individually queue 100 tasks to the worker cluster, which will save their results in the cache backend for speed.
 Once all the 100 results are in the cache, they are collated into a list and saved as a single result in the database. The cache results are then cleared.
+
+You can also use an :class:`Iter` instance which can sometimes be more convenient:
+
+.. code-block:: python
+
+    from django_q.tasks import Iter
+
+    i = Iter('math.copysign')
+
+    # add some arguments
+    i.append(1, -1)
+    i.append(2, -1)
+    i.append(3, -1)
+
+    # run it
+    i.run()
+
+    # get the results
+    print(i.result())
+
+.. code-block:: python
+
+    [-1.0, -2.0, -3.0]
+
 Needs the Django cache framework.
 
 .. _groups:
@@ -179,6 +203,45 @@ You can also access group functions from a task result instance:
         task.group_delete()
         print('Deleted group {}'.format(task.group))
 
+Chains
+------
+Sometimes you want to run tasks sequentially. For that you can use the :func:`async_chain` function:
+
+.. code-block:: python
+
+    # Async a chain of tasks
+    from django_q.tasks import async_chain, result_group
+
+    # the chain must be in the format
+    # [(func,(args),{kwargs}),(func,(args),{kwargs}),..]
+    group_id = async_chain([('math.copysign', (1, -1)),
+                            ('math.floor', (1,))])
+
+    # get group result
+    result_group(group_id, count=2)
+
+A slightly more convenient way is to use a :class:`Chain` instance:
+
+.. code-block:: python
+
+    # Chain async
+    from django_q.tasks import Chain
+
+    # create a chain that uses the cache backend
+    chain = Chain(cached=True)
+
+    # add some tasks
+    chain.append('math.copysign', 1, -1)
+    chain.append('math.floor', 1)
+
+    # run it
+    chain.run()
+
+    print(chain.result())
+.. code-block:: python
+
+    [-1.0, 1]
+
 Cached operations
 -----------------
 You can run your tasks results against the Django cache backend instead of the database backend by either using the global :ref:`cached` setting or by supplying the ``cached`` keyword to individual functions.
@@ -323,7 +386,7 @@ Reference
 .. py:function:: async_iter(func, args_iter,**kwargs)
 
    Runs iterable arguments against the cache backend and returns a single collated result.
-   Accepts the same options as :func:`async` except ``hook``.
+   Accepts the same options as :func:`async` except ``hook``. See also the :class:`Iter` class.
 
    :param object func: The task function to execute
    :param args: An iterable containing arguments for the task function
@@ -331,6 +394,17 @@ Reference
    :returns: The uuid of the task
    :rtype: str
 
+
+.. py:function:: async_chain(chain, group=None, cached=Conf.CACHED, sync=Conf.SYNC, broker=None)
+
+    Async a chain of tasks. See also the :class:`Chain` class.
+
+    :param list chain: a list of tasks in the format [(func,(args),{kwargs}), (func,(args),{kwargs})]
+    :param str group: an optional group name.
+    :param bool cached: run this against the cache backend
+    :param bool sync: execute this inline instead of asynchronous
+
+
 .. py:function:: queue_size()
 
     Returns the size of the broker queue.
@@ -499,4 +573,111 @@ Reference
 
 .. py:class:: Failure
 
-     A proxy model of :class:`Task` with the queryset filtered on :attr:`Task.success` is ``False``.
+     A proxy model of :class:`Task` with the queryset filtered on :attr:`Task.success` is ``False``.
+
+
+.. py:class:: Iter(func=None, args=None, kwargs=None, cached=Conf.CACHED, sync=Conf.SYNC, broker=None)
+
+    An async task with iterable arguments. Serves as a convenient wrapper for :func:`async_iter`
+    You can pass the iterable arguments at construction or you can append individual argument tuples.
+
+        :param func: the function to execute
+        :param args: an iterable of arguments.
+        :param kwargs: the keyword arguments
+        :param bool cached: run this against the cache backend
+        :param bool sync: execute this inline instead of asynchronous
+        :param broker: optional broker instance
+
+
+    .. py:method:: append(*args)
+
+    Append arguments to the iter set. Returns the current set count.
+
+        :param args: the arguments for a single execution
+        :return: the current set count
+        :rtype: int
+
+
+    .. py:method:: run()
+
+    Start queueing the tasks to the worker cluster.
+
+        :return: the task result id
+
+
+    .. py:method:: result(wait=0)
+
+    return the full list of results.
+
+        :param int wait: how many milliseconds to wait for a result
+        :return: an unsorted list of results
+
+
+    .. py:method:: fetch(wait=0)
+
+    get the task result objects.
+
+        :param int wait: how many milliseconds to wait for a result
+        :return: an unsorted list of task objects
+
+
+    .. py:method:: length()
+
+    get the length of the arguments list
+
+        :return int: length of the argument list
+
+
+.. py:class:: Chain(chain=None, group=None, cached=Conf.CACHED, sync=Conf.SYNC)
+
+    A sequential chain of tasks. Acts as a convenient wrapper for :func:`async_chain`
+    You can pass the task chain at construction or you can append individual tasks before running them.
+
+        :param list chain: a list of task in the format [(func,(args),{kwargs}), (func,(args),{kwargs})]
+        :param str group: an optional group name.
+        :param bool cached: run this against the cache backend
+        :param bool sync: execute this inline instead of asynchronous
+
+
+    .. py:method:: append(func, *args, **kwargs)
+
+    Append a task to the chain. Takes the same arguments as :func:`async`
+
+        :return: the current number of tasks in the chain
+        :rtype: int
+
+
+    .. py:method:: run()
+
+    Start queueing the chain to the worker cluster.
+
+        :return: the chains group id
+
+
+    .. py:method:: result(wait=0)
+
+    return the full list of results from the chain when it finishes. Blocks until timeout or result.
+
+        :param int wait: how many milliseconds to wait for a result
+        :return: an unsorted list of results
+
+
+    .. py:method:: fetch(failures=True, wait=0)
+
+    get the task result objects from the chain when it finishes. Blocks until timeout or result.
+
+        :param failures: include failed tasks
+        :param int wait: how many milliseconds to wait for a result
+        :return: an unsorted list of task objects
+
+    .. py:method:: current()
+
+    get the index of the currently executing chain element
+
+        :return int: current chain index
+
+    .. py:method:: length()
+
+    get the length of the chain
+
+        :return int: length of the chain