From 33282cb2e53bae22d39e51f025b8db0c135ee135 Mon Sep 17 00:00:00 2001
From: Ilan Steemers <koed00@gmail.com>
Date: Thu, 1 Oct 2015 14:58:28 +0200
Subject: [PATCH 01/14] cached result backend

first version of a result backend using django's cache framework
---
 django_q/cluster.py           |  28 +++++++-
 django_q/conf.py              |   4 ++
 django_q/tasks.py             | 128 ++++++++++++++++++++++++++++++++--
 django_q/tests/test_cached.py |  52 ++++++++++++++
 4 files changed, 204 insertions(+), 8 deletions(-)
 create mode 100644 django_q/tests/test_cached.py

diff --git a/django_q/cluster.py b/django_q/cluster.py
index 343c5ca..8694294 100644
--- a/django_q/cluster.py
+++ b/django_q/cluster.py
@@ -325,7 +325,11 @@ def monitor(result_queue, broker=None):
         if ack_id:
             broker.acknowledge(ack_id)
         # save the result
-        save_task(task)
+        if task.get('cached', False):
+            save_cache(task, broker)
+        else:
+            save_task(task)
+        # log the result
         if task['success']:
             logger.info(_("Processed [{}]").format(task['name']))
         else:
@@ -384,7 +388,7 @@ def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT):
 
 def save_task(task):
     """
-    Saves the task package to Django
+    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']:
@@ -409,6 +413,26 @@ def save_task(task):
         logger.error(e)
 
 
+def save_cache(task, broker):
+    key = 'django_q:{}:results'.format(broker.list_key)
+    timeout = task['cached']
+    if timeout is True:
+        timeout = None
+    try:
+        task_package = signing.SignedPackage.dumps(task)
+        group = task.get('group', False)
+        if group:
+            group_list = broker.cache.get('{}:{}'.format(key, group)) or []
+            group_list.append(task_package)
+            broker.cache.set('{}:{}'.format(key, group), group_list, timeout)
+        else:
+            broker.cache.set('{}:{}'.format(key, task['id']),
+                             task_package,
+                             timeout)
+    except Exception as e:
+        logger.error(e)
+
+
 def scheduler(broker=None):
     """
     Creates a task from a schedule at the scheduled time and schedules next run
diff --git a/django_q/conf.py b/django_q/conf.py
index 9ca2146..ccfd080 100644
--- a/django_q/conf.py
+++ b/django_q/conf.py
@@ -113,6 +113,10 @@ class Conf(object):
     # The Django cache to use
     CACHE = conf.get('cache', 'default')
 
+    # Use the cache as result backend. Can be 'True' or an integer representing the global cache timeout.
+    # i.e 'cached: 60' , will make all results go the cache and expire in 60 seconds.
+    CACHED = conf.get('cached', False)
+
     # If set to False the scheduler won't execute tasks in the past.
     # Instead it will run once and reschedule the next run in the future. Defaults to True.
     CATCH_UP = conf.get('catch_up', True)
diff --git a/django_q/tasks.py b/django_q/tasks.py
index 62da8f1..cc5a13b 100644
--- a/django_q/tasks.py
+++ b/django_q/tasks.py
@@ -23,6 +23,7 @@ def async(func, *args, **kwargs):
     sync = options.pop('sync', False)
     group = options.pop('group', None)
     save = options.pop('save', None)
+    cached = options.pop('cached', Conf.CACHED)
     # get an id
     tag = uuid()
     # build the task package
@@ -38,6 +39,8 @@ def async(func, *args, **kwargs):
         task['group'] = group
     if save is not None:
         task['save'] = save
+    if cached:
+        task['cached'] = cached
     # sign it
     pack = signing.SignedPackage.dumps(task)
     if sync or Conf.SYNC:
@@ -83,7 +86,7 @@ def schedule(func, *args, **kwargs):
                                    )
 
 
-def result(task_id, wait=0):
+def result(task_id, wait=0, cached=Conf.CACHED):
     """
     Return the result of the named task.
 
@@ -94,6 +97,8 @@ def result(task_id, wait=0):
     :return: the result object of this task
     :rtype: object
     """
+    if cached:
+        return result_cached(task_id, wait)
     start = time.time()
     while True:
         r = Task.get_result(task_id)
@@ -104,7 +109,21 @@ def result(task_id, wait=0):
         time.sleep(0.01)
 
 
-def result_group(group_id, failures=False):
+def result_cached(task_id, wait=0, broker=None):
+    if not broker:
+        broker = get_broker()
+    key = 'django_q:{}:results'.format(broker.list_key)
+    start = time.time()
+    while True:
+        r = broker.cache.get('{}:{}'.format(key, task_id))
+        if r:
+            return signing.SignedPackage.loads(r)['result']
+        if (time.time() - start) * 1000 >= wait:
+            break
+        time.sleep(0.01)
+
+
+def result_group(group_id, failures=False, cached=Conf.CACHED):
     """
     Return a list of results for a task group.
 
@@ -112,10 +131,26 @@ def result_group(group_id, failures=False):
     :param bool failures: set to True to include failures
     :return: list or results
     """
+    if cached:
+        return result_group_cached(group_id, failures)
     return Task.get_result_group(group_id, failures)
 
 
-def fetch(task_id, wait=0):
+def result_group_cached(group_id, failures=False, broker=None):
+    if not broker:
+        broker = get_broker()
+    key = 'django_q:{}:results'.format(broker.list_key)
+    group_list = broker.cache.get('{}:{}'.format(key, group_id))
+    if group_list:
+        result_list = []
+        for task_package in group_list:
+            task = signing.SignedPackage.loads(task_package)
+            if task['success'] or failures:
+                result_list.append(task['result'])
+        return result_list
+
+
+def fetch(task_id, wait=0, cached=Conf.CACHED):
     """
     Return the processed task.
 
@@ -126,6 +161,8 @@ def fetch(task_id, wait=0):
     :return: the full task object
     :rtype: Task
     """
+    if cached:
+        return fetch_cached(task_id, wait)
     start = time.time()
     while True:
         t = Task.get_task(task_id)
@@ -136,7 +173,32 @@ def fetch(task_id, wait=0):
         time.sleep(0.01)
 
 
-def fetch_group(group_id, failures=True):
+def fetch_cached(task_id, wait=0, broker=None):
+    if not broker:
+        broker = get_broker()
+    key = 'django_q:{}:results'.format(broker.list_key)
+    start = time.time()
+    while True:
+        r = broker.cache.get('{}:{}'.format(key, task_id))
+        if r:
+            task = signing.SignedPackage.loads(r)
+            t = Task(id=task['id'],
+                     name=task['name'],
+                     func=task['func'],
+                     hook=task.get('hook'),
+                     args=task['args'],
+                     kwargs=task['kwargs'],
+                     started=task['started'],
+                     stopped=task['stopped'],
+                     result=task['result'],
+                     success=task['success'])
+            return t
+        if (time.time() - start) * 1000 >= wait:
+            break
+        time.sleep(0.01)
+
+
+def fetch_group(group_id, failures=True, cached=True):
     """
     Return a list of Tasks for a task group.
 
@@ -144,10 +206,37 @@ def fetch_group(group_id, failures=True):
     :param bool failures: set to False to exclude failures
     :return: list of Tasks
     """
+    if cached:
+        return fetch_group_cached(group_id, failures)
     return Task.get_task_group(group_id, failures)
 
 
-def count_group(group_id, failures=False):
+def fetch_group_cached(group_id, failures=True, broker=None):
+    if not broker:
+        broker = get_broker()
+    key = 'django_q:{}:results'.format(broker.list_key)
+    group_list = broker.cache.get('{}:{}'.format(key, group_id))
+    if group_list:
+        task_list = []
+        for task_package in group_list:
+            task = signing.SignedPackage.loads(task_package)
+            if task['success'] or failures:
+                t = Task(id=task['id'],
+                         name=task['name'],
+                         func=task['func'],
+                         hook=task.get('hook'),
+                         args=task['args'],
+                         kwargs=task['kwargs'],
+                         started=task['started'],
+                         stopped=task['stopped'],
+                         result=task['result'],
+                         group=task.get('group'),
+                         success=task['success'])
+                task_list.append(t)
+        return task_list
+
+
+def count_group(group_id, failures=False, cached=Conf.CACHED):
     """
     Count the results in a group.
 
@@ -156,10 +245,28 @@ def count_group(group_id, failures=False):
     :return: the number of tasks/results in a group
     :rtype: int
     """
+    if cached:
+        return count_group_cached(group_id, failures)
     return Task.get_group_count(group_id, failures)
 
 
-def delete_group(group_id, tasks=False):
+def count_group_cached(group_id, failures=False, broker=None):
+    if not broker:
+        broker = get_broker()
+    key = 'django_q:{}:results'.format(broker.list_key)
+    group_list = broker.cache.get('{}:{}'.format(key, group_id))
+    if group_list:
+        if not failures:
+            return len(group_list)
+        failure_count = 0
+        for task_package in group_list:
+            task = signing.SignedPackage.loads(task_package)
+            if not task['success']:
+                failure_count += 1
+        return failure_count
+
+
+def delete_group(group_id, tasks=False, cached=Conf.CACHE):
     """
     Delete a group.
 
@@ -168,9 +275,18 @@ def delete_group(group_id, tasks=False):
     Otherwise just the group label is removed.
     :return:
     """
+    if cached:
+        return delete_group_cached(group_id)
     return Task.delete_group(group_id, tasks)
 
 
+def delete_group_cached(group_id, broker=None):
+    if not broker:
+        broker = get_broker()
+    key = 'django_q:{}:results'.format(broker.list_key)
+    return broker.cache.delete('{}:{}'.format(key, group_id))
+
+
 def queue_size(broker=None):
     """
     Returns the current queue size.
diff --git a/django_q/tests/test_cached.py b/django_q/tests/test_cached.py
new file mode 100644
index 0000000..ba44d11
--- /dev/null
+++ b/django_q/tests/test_cached.py
@@ -0,0 +1,52 @@
+from multiprocessing import Event
+
+import pytest
+
+from django_q.cluster import Sentinel
+from django_q.conf import Conf
+from django_q.tasks import async, result, fetch, count_group, result_group, fetch_group, delete_group
+from django_q.brokers import get_broker
+
+
+@pytest.fixture
+def broker():
+    Conf.DISQUE_NODES = None
+    Conf.IRON_MQ = None
+    Conf.SQS = None
+    Conf.ORM = None
+    Conf.MONGO = None
+    Conf.DJANGO_REDIS = 'default'
+    return get_broker()
+
+
+@pytest.mark.django_db
+def test_cached(broker):
+    broker.purge_queue()
+    broker.cache.clear()
+    group = 'cache_test'
+    # queue the tests
+    task_id = async('math.copysign', 1, -1, cached=True, broker=broker)
+    async('math.copysign', 1, -1, cached=True, broker=broker, group=group)
+    async('math.copysign', 1, -1, cached=True, broker=broker, group=group)
+    async('math.copysign', 1, -1, cached=True, broker=broker, group=group)
+    async('math.copysign', 1, -1, cached=True, broker=broker, group=group)
+    async('math.copysign', 1, -1, cached=True, broker=broker, group=group)
+    # run a single cluster
+    start_event = Event()
+    stop_event = Event()
+    stop_event.set()
+    Sentinel(stop_event, start_event, broker=broker)
+    # assert results
+    assert result(task_id, wait=500, cached=True) == -1
+    assert fetch(task_id,wait=500, cached=True).result == -1
+    # make sure it's not in the db backend
+    assert fetch(task_id) is None
+    # assert group
+    assert count_group(group, cached=True) == 5
+    assert count_group(group, cached=True, failures=True) == 0
+    assert result_group(group, cached=True) == [-1, -1, -1, -1, -1]
+    assert len(fetch_group(group, cached=True)) == 5
+    assert len(fetch_group(group, cached=True, failures=False)) == 5
+    delete_group(group, cached=True)
+    assert count_group(group, cached=True) is None
+    broker.cache.clear()

From 7107e9ea3565aad913c1e47b6f28527de0a852c5 Mon Sep 17 00:00:00 2001
From: Ilan Steemers <koed00@gmail.com>
Date: Thu, 1 Oct 2015 15:06:43 +0200
Subject: [PATCH 02/14] fetch group should default to non cached

---
 django_q/tasks.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/django_q/tasks.py b/django_q/tasks.py
index cc5a13b..25d8e8c 100644
--- a/django_q/tasks.py
+++ b/django_q/tasks.py
@@ -198,7 +198,7 @@ def fetch_cached(task_id, wait=0, broker=None):
         time.sleep(0.01)
 
 
-def fetch_group(group_id, failures=True, cached=True):
+def fetch_group(group_id, failures=True, cached=False):
     """
     Return a list of Tasks for a task group.
 

From 6127b551aefe5f807570f31b97d1aa6a8c1488a3 Mon Sep 17 00:00:00 2001
From: Ilan Steemers <koed00@gmail.com>
Date: Thu, 1 Oct 2015 17:21:26 +0200
Subject: [PATCH 03/14] fixes tests for cache with better coverage

---
 django_q/tasks.py             |  2 +-
 django_q/tests/test_cached.py | 11 ++++++++---
 2 files changed, 9 insertions(+), 4 deletions(-)

diff --git a/django_q/tasks.py b/django_q/tasks.py
index 25d8e8c..20e8af3 100644
--- a/django_q/tasks.py
+++ b/django_q/tasks.py
@@ -266,7 +266,7 @@ def count_group_cached(group_id, failures=False, broker=None):
         return failure_count
 
 
-def delete_group(group_id, tasks=False, cached=Conf.CACHE):
+def delete_group(group_id, tasks=False, cached=Conf.CACHED):
     """
     Delete a group.
 
diff --git a/django_q/tests/test_cached.py b/django_q/tests/test_cached.py
index ba44d11..cfc0934 100644
--- a/django_q/tests/test_cached.py
+++ b/django_q/tests/test_cached.py
@@ -31,6 +31,10 @@ def test_cached(broker):
     async('math.copysign', 1, -1, cached=True, broker=broker, group=group)
     async('math.copysign', 1, -1, cached=True, broker=broker, group=group)
     async('math.copysign', 1, -1, cached=True, broker=broker, group=group)
+    async('math.popysign', 1, -1, cached=True, broker=broker, group=group)
+    # test wait on cache
+    assert result(task_id, wait=1, cached=True) is None
+    assert fetch(task_id, wait=1, cached=True) is None
     # run a single cluster
     start_event = Event()
     stop_event = Event()
@@ -42,10 +46,11 @@ def test_cached(broker):
     # make sure it's not in the db backend
     assert fetch(task_id) is None
     # assert group
-    assert count_group(group, cached=True) == 5
-    assert count_group(group, cached=True, failures=True) == 0
+    assert count_group(group, cached=True) == 6
+    assert count_group(group, cached=True, failures=True) == 1
     assert result_group(group, cached=True) == [-1, -1, -1, -1, -1]
-    assert len(fetch_group(group, cached=True)) == 5
+    assert len(result_group(group, cached=True, failures=True)) == 6
+    assert len(fetch_group(group, cached=True)) == 6
     assert len(fetch_group(group, cached=True, failures=False)) == 5
     delete_group(group, cached=True)
     assert count_group(group, cached=True) is None

From f6d14bc7b38c1b64a9efd1c59629078f86ad80b9 Mon Sep 17 00:00:00 2001
From: Ilan Steemers <koed00@gmail.com>
Date: Thu, 1 Oct 2015 19:25:43 +0200
Subject: [PATCH 04/14] adds delete_cached

---
 django_q/tasks.py             | 8 +++++++-
 django_q/tests/test_cached.py | 5 ++++-
 2 files changed, 11 insertions(+), 2 deletions(-)

diff --git a/django_q/tasks.py b/django_q/tasks.py
index 20e8af3..ab22053 100644
--- a/django_q/tasks.py
+++ b/django_q/tasks.py
@@ -281,10 +281,16 @@ def delete_group(group_id, tasks=False, cached=Conf.CACHED):
 
 
 def delete_group_cached(group_id, broker=None):
+    if not broker:
+        broker = get_broker()
+    return delete_cached(group_id, broker)
+
+
+def delete_cached(task_id, broker=None):
     if not broker:
         broker = get_broker()
     key = 'django_q:{}:results'.format(broker.list_key)
-    return broker.cache.delete('{}:{}'.format(key, group_id))
+    return broker.cache.delete('{}:{}'.format(key, task_id))
 
 
 def queue_size(broker=None):
diff --git a/django_q/tests/test_cached.py b/django_q/tests/test_cached.py
index cfc0934..9a9f879 100644
--- a/django_q/tests/test_cached.py
+++ b/django_q/tests/test_cached.py
@@ -4,7 +4,7 @@ import pytest
 
 from django_q.cluster import Sentinel
 from django_q.conf import Conf
-from django_q.tasks import async, result, fetch, count_group, result_group, fetch_group, delete_group
+from django_q.tasks import async, result, fetch, count_group, result_group, fetch_group, delete_group, delete_cached
 from django_q.brokers import get_broker
 
 
@@ -54,4 +54,7 @@ def test_cached(broker):
     assert len(fetch_group(group, cached=True, failures=False)) == 5
     delete_group(group, cached=True)
     assert count_group(group, cached=True) is None
+    delete_cached(task_id)
+    assert result(task_id, cached=True) is None
+    assert fetch(task_id, cached=True) is None
     broker.cache.clear()

From bcdb62404b7dc031a1fb1b0c6b1327633ba6440e Mon Sep 17 00:00:00 2001
From: Ilan Steemers <koed00@gmail.com>
Date: Fri, 2 Oct 2015 13:16:13 +0200
Subject: [PATCH 05/14] updated botocore and dependency

---
 requirements.txt | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/requirements.txt b/requirements.txt
index 65e0154..a331cc9 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -7,7 +7,7 @@
 arrow==0.6.0
 blessed==1.9.5
 boto3==1.1.4
-botocore==1.2.6           # via boto3
+botocore==1.2.7           # via boto3
 django-picklefield==0.3.2
 django-redis==4.2.0
 docutils==0.12            # via botocore
@@ -16,7 +16,7 @@ futures==2.2.0            # via boto3
 hiredis==0.2.0
 iron-core==1.1.9          # via iron-mq
 iron-mq==0.7
-jmespath==0.8.0           # via boto3, botocore
+jmespath==0.9.0           # via boto3, botocore
 msgpack-python==0.4.6     # via django-redis
 psutil==3.2.1
 pymongo==3.0.3

From 9cfc41f1cb26bb16d06654c32f6ac78eced33f72 Mon Sep 17 00:00:00 2001
From: Ilan Steemers <koed00@gmail.com>
Date: Fri, 2 Oct 2015 16:40:31 +0200
Subject: [PATCH 06/14] fixed global cached for fetch group

And added docstrings
---
 django_q/tasks.py | 31 +++++++++++++++++++++++++++++--
 1 file changed, 29 insertions(+), 2 deletions(-)

diff --git a/django_q/tasks.py b/django_q/tasks.py
index ab22053..92207c8 100644
--- a/django_q/tasks.py
+++ b/django_q/tasks.py
@@ -1,4 +1,4 @@
-"""Provides task functionalities."""
+"""Provides task functionality."""
 from multiprocessing import Queue, Value
 
 # django
@@ -94,6 +94,7 @@ def result(task_id, wait=0, cached=Conf.CACHED):
     :param task_id: the task name or uuid
     :type wait: int
     :param wait: number of milliseconds to wait for a result
+    :param cached: run this against the cache backend
     :return: the result object of this task
     :rtype: object
     """
@@ -110,6 +111,9 @@ def result(task_id, wait=0, cached=Conf.CACHED):
 
 
 def result_cached(task_id, wait=0, broker=None):
+    """
+     Return the result from the cache backend
+    """
     if not broker:
         broker = get_broker()
     key = 'django_q:{}:results'.format(broker.list_key)
@@ -129,6 +133,7 @@ def result_group(group_id, failures=False, cached=Conf.CACHED):
 
     :param str group_id: the group id
     :param bool failures: set to True to include failures
+    :param cached: run this against the cache backend
     :return: list or results
     """
     if cached:
@@ -137,6 +142,9 @@ def result_group(group_id, failures=False, cached=Conf.CACHED):
 
 
 def result_group_cached(group_id, failures=False, broker=None):
+    """
+    Return a list of results for a task group from the cache backend
+    """
     if not broker:
         broker = get_broker()
     key = 'django_q:{}:results'.format(broker.list_key)
@@ -158,6 +166,7 @@ def fetch(task_id, wait=0, cached=Conf.CACHED):
     :type task_id: str or uuid
     :param wait: the number of milliseconds to wait for a result
     :type wait: int
+    :param cached: run this against the cache backend
     :return: the full task object
     :rtype: Task
     """
@@ -174,6 +183,9 @@ def fetch(task_id, wait=0, cached=Conf.CACHED):
 
 
 def fetch_cached(task_id, wait=0, broker=None):
+    """
+    Return the processed task from the cache backend
+    """
     if not broker:
         broker = get_broker()
     key = 'django_q:{}:results'.format(broker.list_key)
@@ -198,12 +210,13 @@ def fetch_cached(task_id, wait=0, broker=None):
         time.sleep(0.01)
 
 
-def fetch_group(group_id, failures=True, cached=False):
+def fetch_group(group_id, failures=True, cached=Conf.CACHED):
     """
     Return a list of Tasks for a task group.
 
     :param str group_id: the group id
     :param bool failures: set to False to exclude failures
+    :param cached: run this against the cache backend
     :return: list of Tasks
     """
     if cached:
@@ -212,6 +225,9 @@ def fetch_group(group_id, failures=True, cached=False):
 
 
 def fetch_group_cached(group_id, failures=True, broker=None):
+    """
+    Return a list of Tasks for a task group in the cache backend
+    """
     if not broker:
         broker = get_broker()
     key = 'django_q:{}:results'.format(broker.list_key)
@@ -242,6 +258,7 @@ def count_group(group_id, failures=False, cached=Conf.CACHED):
 
     :param str group_id: the group id
     :param bool failures: Returns failure count if True
+    :param cached: run this against the cache backend
     :return: the number of tasks/results in a group
     :rtype: int
     """
@@ -251,6 +268,9 @@ def count_group(group_id, failures=False, cached=Conf.CACHED):
 
 
 def count_group_cached(group_id, failures=False, broker=None):
+    """
+    Count the results in a group in the cache backend
+    """
     if not broker:
         broker = get_broker()
     key = 'django_q:{}:results'.format(broker.list_key)
@@ -273,6 +293,7 @@ def delete_group(group_id, tasks=False, cached=Conf.CACHED):
     :param str group_id: the group id
     :param bool tasks: If set to True this will also delete the group tasks.
     Otherwise just the group label is removed.
+    :param cached: run this against the cache backend
     :return:
     """
     if cached:
@@ -281,12 +302,18 @@ def delete_group(group_id, tasks=False, cached=Conf.CACHED):
 
 
 def delete_group_cached(group_id, broker=None):
+    """
+    Delete a group from the cache backend
+    """
     if not broker:
         broker = get_broker()
     return delete_cached(group_id, broker)
 
 
 def delete_cached(task_id, broker=None):
+    """
+    Delete a task from the cache backend
+    """
     if not broker:
         broker = get_broker()
     key = 'django_q:{}:results'.format(broker.list_key)

From 81d5abbd00b0bbe1f45335082aa33eca86c98d2a Mon Sep 17 00:00:00 2001
From: Ilan Steemers <koed00@gmail.com>
Date: Fri, 2 Oct 2015 18:29:03 +0200
Subject: [PATCH 07/14] adds count  and wait option to group result and fetch

result_group and fetch_group can now be told to block for a certain number of results or a number of miliseconds or a combination of both.
---
 django_q/tasks.py | 129 +++++++++++++++++++++++++++++++---------------
 1 file changed, 87 insertions(+), 42 deletions(-)

diff --git a/django_q/tasks.py b/django_q/tasks.py
index 92207c8..331d27c 100644
--- a/django_q/tasks.py
+++ b/django_q/tasks.py
@@ -15,7 +15,7 @@ from django_q.brokers import get_broker
 
 
 def async(func, *args, **kwargs):
-    """Send a task to the cluster."""
+    """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)
@@ -94,7 +94,7 @@ def result(task_id, wait=0, cached=Conf.CACHED):
     :param task_id: the task name or uuid
     :type wait: int
     :param wait: number of milliseconds to wait for a result
-    :param cached: run this against the cache backend
+    :param bool cached: run this against the cache backend
     :return: the result object of this task
     :rtype: object
     """
@@ -127,35 +127,58 @@ def result_cached(task_id, wait=0, broker=None):
         time.sleep(0.01)
 
 
-def result_group(group_id, failures=False, cached=Conf.CACHED):
+def result_group(group_id, failures=False, wait=0, count=None, cached=Conf.CACHED):
     """
     Return a list of results for a task group.
 
     :param str group_id: the group id
     :param bool failures: set to True to include failures
-    :param cached: run this against the cache backend
+    :param int count: Block until there are this many results in the group
+    :param bool cached: run this against the cache backend
     :return: list or results
     """
     if cached:
-        return result_group_cached(group_id, failures)
-    return Task.get_result_group(group_id, failures)
+        return result_group_cached(group_id, failures, wait, count)
+    start = time.time()
+    if count:
+        while True:
+            if count_group(group_id) == count or wait and (time.time() - start) * 1000 >= wait:
+                break
+            time.sleep(0.01)
+    while True:
+        r = Task.get_result_group(group_id, failures)
+        if r:
+            return r
+        if (time.time() - start) * 1000 >= wait:
+            break
+        time.sleep(0.01)
 
 
-def result_group_cached(group_id, failures=False, broker=None):
+def result_group_cached(group_id, failures=False, wait=0, count=None, broker=None):
     """
     Return a list of results for a task group from the cache backend
     """
     if not broker:
         broker = get_broker()
+    start = time.time()
+    if count:
+        while True:
+            if count_group_cached(group_id) == count or wait and (time.time() - start) * 1000 >= wait:
+                break
+            time.sleep(0.01)
     key = 'django_q:{}:results'.format(broker.list_key)
-    group_list = broker.cache.get('{}:{}'.format(key, group_id))
-    if group_list:
-        result_list = []
-        for task_package in group_list:
-            task = signing.SignedPackage.loads(task_package)
-            if task['success'] or failures:
-                result_list.append(task['result'])
-        return result_list
+    while True:
+        group_list = broker.cache.get('{}:{}'.format(key, group_id))
+        if group_list:
+            result_list = []
+            for task_package in group_list:
+                task = signing.SignedPackage.loads(task_package)
+                if task['success'] or failures:
+                    result_list.append(task['result'])
+            return result_list
+        if (time.time() - start) * 1000 >= wait:
+            break
+        time.sleep(0.01)
 
 
 def fetch(task_id, wait=0, cached=Conf.CACHED):
@@ -166,7 +189,7 @@ def fetch(task_id, wait=0, cached=Conf.CACHED):
     :type task_id: str or uuid
     :param wait: the number of milliseconds to wait for a result
     :type wait: int
-    :param cached: run this against the cache backend
+    :param bool cached: run this against the cache backend
     :return: the full task object
     :rtype: Task
     """
@@ -210,46 +233,68 @@ def fetch_cached(task_id, wait=0, broker=None):
         time.sleep(0.01)
 
 
-def fetch_group(group_id, failures=True, cached=Conf.CACHED):
+def fetch_group(group_id, failures=True, wait=0, count=None, cached=Conf.CACHED):
     """
     Return a list of Tasks for a task group.
 
     :param str group_id: the group id
     :param bool failures: set to False to exclude failures
-    :param cached: run this against the cache backend
+    :param bool cached: run this against the cache backend
     :return: list of Tasks
     """
     if cached:
-        return fetch_group_cached(group_id, failures)
-    return Task.get_task_group(group_id, failures)
+        return fetch_group_cached(group_id, failures, wait, count)
+    start = time.time()
+    if count:
+        while True:
+            if count_group(group_id) == count or wait and (time.time() - start) * 1000 >= wait:
+                break
+            time.sleep(0.01)
+    while True:
+        r = Task.get_task_group(group_id, failures)
+        if r:
+            return r
+        if (time.time() - start) * 1000 >= wait:
+            break
+        time.sleep(0.01)
 
 
-def fetch_group_cached(group_id, failures=True, broker=None):
+def fetch_group_cached(group_id, failures=True, wait=0, count=None, broker=None):
     """
     Return a list of Tasks for a task group in the cache backend
     """
     if not broker:
         broker = get_broker()
+    start = time.time()
+    if count:
+        while True:
+            if count_group_cached(group_id) == count or wait and (time.time() - start) * 1000 >= wait:
+                break
+            time.sleep(0.01)
     key = 'django_q:{}:results'.format(broker.list_key)
-    group_list = broker.cache.get('{}:{}'.format(key, group_id))
-    if group_list:
-        task_list = []
-        for task_package in group_list:
-            task = signing.SignedPackage.loads(task_package)
-            if task['success'] or failures:
-                t = Task(id=task['id'],
-                         name=task['name'],
-                         func=task['func'],
-                         hook=task.get('hook'),
-                         args=task['args'],
-                         kwargs=task['kwargs'],
-                         started=task['started'],
-                         stopped=task['stopped'],
-                         result=task['result'],
-                         group=task.get('group'),
-                         success=task['success'])
-                task_list.append(t)
-        return task_list
+    while True:
+        group_list = broker.cache.get('{}:{}'.format(key, group_id))
+        if group_list:
+            task_list = []
+            for task_package in group_list:
+                task = signing.SignedPackage.loads(task_package)
+                if task['success'] or failures:
+                    t = Task(id=task['id'],
+                             name=task['name'],
+                             func=task['func'],
+                             hook=task.get('hook'),
+                             args=task['args'],
+                             kwargs=task['kwargs'],
+                             started=task['started'],
+                             stopped=task['stopped'],
+                             result=task['result'],
+                             group=task.get('group'),
+                             success=task['success'])
+                    task_list.append(t)
+            return task_list
+        if (time.time() - start) * 1000 >= wait:
+            break
+        time.sleep(0.01)
 
 
 def count_group(group_id, failures=False, cached=Conf.CACHED):
@@ -258,7 +303,7 @@ def count_group(group_id, failures=False, cached=Conf.CACHED):
 
     :param str group_id: the group id
     :param bool failures: Returns failure count if True
-    :param cached: run this against the cache backend
+    :param bool cached: run this against the cache backend
     :return: the number of tasks/results in a group
     :rtype: int
     """
@@ -293,7 +338,7 @@ def delete_group(group_id, tasks=False, cached=Conf.CACHED):
     :param str group_id: the group id
     :param bool tasks: If set to True this will also delete the group tasks.
     Otherwise just the group label is removed.
-    :param cached: run this against the cache backend
+    :param bool cached: run this against the cache backend
     :return:
     """
     if cached:

From 7911419642909025ad169c9f76ccceaa0219e2dc Mon Sep 17 00:00:00 2001
From: Ilan Steemers <koed00@gmail.com>
Date: Fri, 2 Oct 2015 18:31:52 +0200
Subject: [PATCH 08/14] docs: updated cache and group operations

---
 README.rst         |  2 +-
 docs/configure.rst |  7 ++++
 docs/index.rst     |  2 +-
 docs/tasks.rst     | 82 +++++++++++++++++++++++++++++++++++++++++-----
 4 files changed, 83 insertions(+), 10 deletions(-)

diff --git a/README.rst b/README.rst
index eef6594..48edf45 100644
--- a/README.rst
+++ b/README.rst
@@ -15,7 +15,7 @@ Features
 -  Asynchronous tasks
 -  Scheduled and repeated tasks
 -  Encrypted and compressed packages
--  Failure and success database
+-  Failure and success database or cache
 -  Result hooks and groups
 -  Django Admin integration
 -  PaaS compatible with multiple instances
diff --git a/docs/configure.rst b/docs/configure.rst
index 1795670..9f07d4b 100644
--- a/docs/configure.rst
+++ b/docs/configure.rst
@@ -308,6 +308,13 @@ cache
 For some brokers, you will need to set up the Django `cache framework <https://docs.djangoproject.com/en/1.8/topics/cache/#setting-up-the-cache>`__
 to gather statistics for the monitor. You can indicate which cache to use by setting this value. Defaults to ``default``.
 
+.. _cached:
+
+cached
+~~~~~~
+Switches all task and result functions from using the database backend to the cache backend. This is the same as setting the keyword ``cached=True`` on all task functions.
+Instead of a bool this can also be set to the number of seconds you want the cache to retain results. e.g. ``cached=60``
+
 scheduler
 ~~~~~~~~~
 You can disable the scheduler by setting this option to ``False``. This will reduce a little overhead if you're not using schedules, but is most useful if you want to temporarily disable all schedules.
diff --git a/docs/index.rst b/docs/index.rst
index b46ba00..761b8a6 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -15,7 +15,7 @@ Features
 -  Asynchronous tasks
 -  Scheduled and repeated tasks
 -  Encrypted and compressed packages
--  Failure and success database
+-  Failure and success database or cache
 -  Result hooks and groups
 -  Django Admin integration
 -  PaaS compatible with multiple instances
diff --git a/docs/tasks.rst b/docs/tasks.rst
index 5da69c1..1d0efcd 100644
--- a/docs/tasks.rst
+++ b/docs/tasks.rst
@@ -59,6 +59,12 @@ sync
 Simulates a task execution synchronously. Useful for testing.
 Can also be forced globally via the :ref:`sync` configuration option.
 
+cached
+""""""
+Redirects the result to the cache backend instead of the database if set to ``True`` or to an integer indicating the cache timeout in seconds.
+e.g. ``cached=60``. Especially useful with large and group operations where you don't need the all results in your
+database and want to take advantage of the speed of your cache backend.
+
 broker
 """"""
 A broker instance, in case you want to control your own connections.
@@ -149,6 +155,47 @@ You can also access group functions from a task result instance:
         task.group_delete()
         print('Deleted group {}'.format(task.group))
 
+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.
+This can be useful if you are not interested in persistent results or if you run large group tasks where you only want the final result.
+By using a cache backend like Redis or Memcached you can speed up access to your task results significantly compared to a relational database.
+
+When you set ``cached=True``, results will be saved permanently in the cache and you will have to rely on your backend's cleanup strategies (like LRU) to
+manage stale results.
+You can however opt to set a manual timeout on the results, by setting ``cached=60``. Meaning the result will be evicted from the cache after 60 seconds.
+This works both globally or on individual async executions.::
+
+    # simple cached example
+    from django_q.tasks import async, result
+
+    # cache the result for 10 seconds
+    id = async('math.floor', 100, cached=10)
+
+    # wait max 50ms for the result to appear in the cache
+    result(id, wait=50, cached=True)
+
+This also works for group actions::
+
+    # cached group example
+    from django_q.tasks import async, result_group
+    from django_q.brokers import get_broker
+
+    # set up a broker instance for better performance
+    broker = get_broker()
+
+    # async a hundred functions under a group label
+    for i in range(100):
+        async('math.frexp',
+              i,
+              group='frexp',
+              cached=True,
+              broker=broker)
+
+    # wait max 50ms for one hundred results to return
+    result_group('frexp', wait=50, count=100, cached=True)
+
+
 Synchronous testing
 -------------------
 
@@ -199,7 +246,7 @@ Reference
 ---------
 
 .. py:function:: async(func, *args, hook=None, group=None, timeout=None,\
-    save=None, sync=False, broker=None, q_options=None, **kwargs)
+    save=None, sync=False, cached=False, broker=None, q_options=None, **kwargs)
 
     Puts a task in the cluster queue
 
@@ -210,26 +257,29 @@ Reference
    :param int timeout: Overrides global cluster :ref:`timeout`.
    :param bool save: Overrides global save setting for this task.
    :param bool sync: If set to True, async will simulate a task execution
+   :param cached: Output the result to the cache backend. Bool or timeout in seconds
    :param broker: Optional broker connection from :func:`brokers.get_broker`
    :param dict q_options: Options dict, overrides option keywords
    :param dict kwargs: Keyword arguments for the task function
    :returns: The uuid of the task
    :rtype: str
 
-.. py:function:: result(task_id, wait=0)
+.. py:function:: result(task_id, wait=0, cached=False)
 
     Gets the result of a previously executed task
 
     :param str task_id: the uuid or name of the task
     :param int wait: optional milliseconds to wait for a result
+    :param bool cached: run this against the cache backend.
     :returns: The result of the executed task
 
-.. py:function:: fetch(task_id, wait=0)
+.. py:function:: fetch(task_id, wait=0, cached=False)
 
     Returns a previously executed task
 
     :param str name: the uuid or name of the task
     :param int wait: optional milliseconds to wait for a result
+    :param bool cached: run this against the cache backend.
     :returns: A task object
     :rtype: Task
 
@@ -245,42 +295,58 @@ Reference
     :returns: The amount of task packages in the broker
     :rtype: int
 
-.. py:function:: result_group(group_id, failures=False)
+.. py:function:: result_group(group_id, failures=False, wait=0, count=None, cached=False)
 
     Returns the results of a task group
 
     :param str group_id: the group identifier
     :param bool failures: set this to ``True`` to include failed results
+    :param int wait: optional milliseconds to wait for a result or count
+    :param int count: block until there are this many results in the group
+    :param bool cached: run this against the cache backend
     :returns: a list of results
     :rtype: list
 
-.. py:function:: fetch_group(group_id, failures=True)
+.. py:function:: fetch_group(group_id, failures=True, wait=0, count=None, cached=False)
 
     Returns a list of tasks in a group
 
     :param str group_id: the group identifier
     :param bool failures: set this to ``False`` to exclude failed tasks
-    :returns: a list of Tasks
+    :param int wait: optional milliseconds to wait for a task or count
+    :param int count: block until there are this many tasks in the group
+    :param bool cached: run this against the cache backend.
+    :returns: a list of :class:`Task`
     :rtype: list
 
-.. py:function:: count_group(group_id, failures=False)
+.. py:function:: count_group(group_id, failures=False, cached=False)
 
     Counts the number of task results in a group.
 
     :param str group_id: the group identifier
     :param bool failures: counts the number of failures if ``True``
+    :param bool cached: run this against the cache backend.
     :returns: the number of tasks or failures in a group
     :rtype: int
 
-.. py:function:: delete_group(group_id, tasks=False)
+.. py:function:: delete_group(group_id, tasks=False, cached=False)
 
     Deletes a group label from the database.
 
     :param str group_id: the group identifier
     :param bool tasks: also deletes the associated tasks if ``True``
+    :param bool cached: run this against the cache backend.
     :returns: the numbers of tasks affected
     :rtype: int
 
+.. py:function:: delete_cached(task_id, broker=None)
+
+    Deletes a task from the cache backend
+
+    :param task_id: the uuid of the task
+    :param broker: an optional broker instance
+
+
 .. py:class:: Task
 
     Database model describing an executed task

From 9243731a9c8c91e2772d1792e535bd4a1f42bc2d Mon Sep 17 00:00:00 2001
From: Ilan Steemers <koed00@gmail.com>
Date: Fri, 2 Oct 2015 20:03:38 +0200
Subject: [PATCH 09/14] Updates test for new wait and count options

---
 django_q/tests/test_cached.py  | 11 ++++++++---
 django_q/tests/test_cluster.py |  7 +++++++
 2 files changed, 15 insertions(+), 3 deletions(-)

diff --git a/django_q/tests/test_cached.py b/django_q/tests/test_cached.py
index 9a9f879..7e8e618 100644
--- a/django_q/tests/test_cached.py
+++ b/django_q/tests/test_cached.py
@@ -33,8 +33,13 @@ def test_cached(broker):
     async('math.copysign', 1, -1, cached=True, broker=broker, group=group)
     async('math.popysign', 1, -1, cached=True, broker=broker, group=group)
     # test wait on cache
-    assert result(task_id, wait=1, cached=True) is None
-    assert fetch(task_id, wait=1, cached=True) is None
+    # test wait timeout
+    assert result(task_id, wait=10, cached=True) is None
+    assert fetch(task_id, wait=10, cached=True) is None
+    assert result_group(group, wait=10, cached=True) is None
+    assert result_group(group, count=2, wait=10, cached=True) is None
+    assert fetch_group(group, wait=10, cached=True) is None
+    assert fetch_group(group, count=2, wait=10, cached=True) is None
     # run a single cluster
     start_event = Event()
     stop_event = Event()
@@ -42,7 +47,7 @@ def test_cached(broker):
     Sentinel(stop_event, start_event, broker=broker)
     # assert results
     assert result(task_id, wait=500, cached=True) == -1
-    assert fetch(task_id,wait=500, cached=True).result == -1
+    assert fetch(task_id, wait=500, cached=True).result == -1
     # make sure it's not in the db backend
     assert fetch(task_id) is None
     # assert group
diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py
index 37dc79a..5b1350e 100644
--- a/django_q/tests/test_cluster.py
+++ b/django_q/tests/test_cluster.py
@@ -160,6 +160,13 @@ def test_async(broker, admin_user):
     assert broker.queue_size() == 0
     assert task_queue.qsize() == task_count
     task_queue.put('STOP')
+    # test wait timeout
+    assert result(j, wait=10) is None
+    assert fetch(j, wait=10) is None
+    assert result_group('test_j', wait=10) is None
+    assert result_group('test_j', count=2, wait=10) is None
+    assert fetch_group('test_j', wait=10) is None
+    assert fetch_group('test_j', count=2, wait=10) is None
     # let a worker handle them
     result_queue = Queue()
     worker(task_queue, result_queue, Value('f', -1))

From f73ae66c0f2100c0057e216ec37f1d3fe9ce1721 Mon Sep 17 00:00:00 2001
From: Ilan Steemers <koed00@gmail.com>
Date: Sat, 3 Oct 2015 18:07:57 +0200
Subject: [PATCH 10/14] Adds async_iter command

With async iter you can quickly run the same function on an iterable set of arguments. The results are held in the cache until all are done and collated into a database result.
---
 django_q/cluster.py | 39 ++++++++++++++++++++---------
 django_q/tasks.py   | 60 ++++++++++++++++++++++++++++++---------------
 2 files changed, 68 insertions(+), 31 deletions(-)

diff --git a/django_q/cluster.py b/django_q/cluster.py
index 8694294..4ca18f3 100644
--- a/django_q/cluster.py
+++ b/django_q/cluster.py
@@ -326,7 +326,7 @@ def monitor(result_queue, broker=None):
             broker.acknowledge(ack_id)
         # save the result
         if task.get('cached', False):
-            save_cache(task, broker)
+            save_cached(task, broker)
         else:
             save_task(task)
         # log the result
@@ -413,22 +413,39 @@ def save_task(task):
         logger.error(e)
 
 
-def save_cache(task, broker):
-    key = 'django_q:{}:results'.format(broker.list_key)
+def save_cached(task, broker):
+    task_key = '{}:{}'.format(broker.list_key, task['id'])
     timeout = task['cached']
     if timeout is True:
         timeout = None
     try:
-        task_package = signing.SignedPackage.dumps(task)
         group = task.get('group', False)
+        iter_count = task.get('iter_count', None)
+        # if it's a group append to the group list
         if group:
-            group_list = broker.cache.get('{}:{}'.format(key, group)) or []
-            group_list.append(task_package)
-            broker.cache.set('{}:{}'.format(key, group), group_list, timeout)
-        else:
-            broker.cache.set('{}:{}'.format(key, task['id']),
-                             task_package,
-                             timeout)
+            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:
+                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]
+                results.append(task['result'])
+                task['result'] = results
+                task['id'] = group
+                task['args'] = signing.SignedPackage.loads(broker.cache.get(group_args))
+                save_task(task)
+                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)
+        # save the task
+        broker.cache.set(task_key,
+                         signing.SignedPackage.dumps(task),
+                         timeout)
     except Exception as e:
         logger.error(e)
 
diff --git a/django_q/tasks.py b/django_q/tasks.py
index 331d27c..17ae5b1 100644
--- a/django_q/tasks.py
+++ b/django_q/tasks.py
@@ -24,6 +24,7 @@ def async(func, *args, **kwargs):
     group = options.pop('group', None)
     save = options.pop('save', None)
     cached = options.pop('cached', Conf.CACHED)
+    iter_count = options.pop('iter_count', None)
     # get an id
     tag = uuid()
     # build the task package
@@ -41,6 +42,8 @@ def async(func, *args, **kwargs):
         task['save'] = save
     if cached:
         task['cached'] = cached
+    if iter_count:
+        task['iter_count'] = iter_count
     # sign it
     pack = signing.SignedPackage.dumps(task)
     if sync or Conf.SYNC:
@@ -116,10 +119,9 @@ def result_cached(task_id, wait=0, broker=None):
     """
     if not broker:
         broker = get_broker()
-    key = 'django_q:{}:results'.format(broker.list_key)
     start = time.time()
     while True:
-        r = broker.cache.get('{}:{}'.format(key, task_id))
+        r = broker.cache.get('{}:{}'.format(broker.list_key, task_id))
         if r:
             return signing.SignedPackage.loads(r)['result']
         if (time.time() - start) * 1000 >= wait:
@@ -166,13 +168,12 @@ def result_group_cached(group_id, failures=False, wait=0, count=None, broker=Non
             if count_group_cached(group_id) == count or wait and (time.time() - start) * 1000 >= wait:
                 break
             time.sleep(0.01)
-    key = 'django_q:{}:results'.format(broker.list_key)
     while True:
-        group_list = broker.cache.get('{}:{}'.format(key, group_id))
+        group_list = broker.cache.get('{}:{}:keys'.format(broker.list_key, group_id))
         if group_list:
             result_list = []
-            for task_package in group_list:
-                task = signing.SignedPackage.loads(task_package)
+            for task_key in group_list:
+                task = signing.SignedPackage.loads(broker.cache.get(task_key))
                 if task['success'] or failures:
                     result_list.append(task['result'])
             return result_list
@@ -211,10 +212,9 @@ def fetch_cached(task_id, wait=0, broker=None):
     """
     if not broker:
         broker = get_broker()
-    key = 'django_q:{}:results'.format(broker.list_key)
     start = time.time()
     while True:
-        r = broker.cache.get('{}:{}'.format(key, task_id))
+        r = broker.cache.get('{}:{}'.format(broker.list_key, task_id))
         if r:
             task = signing.SignedPackage.loads(r)
             t = Task(id=task['id'],
@@ -271,13 +271,12 @@ def fetch_group_cached(group_id, failures=True, wait=0, count=None, broker=None)
             if count_group_cached(group_id) == count or wait and (time.time() - start) * 1000 >= wait:
                 break
             time.sleep(0.01)
-    key = 'django_q:{}:results'.format(broker.list_key)
     while True:
-        group_list = broker.cache.get('{}:{}'.format(key, group_id))
+        group_list = broker.cache.get('{}:{}:keys'.format(broker.list_key, group_id))
         if group_list:
             task_list = []
-            for task_package in group_list:
-                task = signing.SignedPackage.loads(task_package)
+            for task_key in group_list:
+                task = signing.SignedPackage.loads(broker.cache.get(task_key))
                 if task['success'] or failures:
                     t = Task(id=task['id'],
                              name=task['name'],
@@ -318,14 +317,13 @@ def count_group_cached(group_id, failures=False, broker=None):
     """
     if not broker:
         broker = get_broker()
-    key = 'django_q:{}:results'.format(broker.list_key)
-    group_list = broker.cache.get('{}:{}'.format(key, group_id))
+    group_list = broker.cache.get('{}:{}:keys'.format(broker.list_key, group_id))
     if group_list:
         if not failures:
             return len(group_list)
         failure_count = 0
-        for task_package in group_list:
-            task = signing.SignedPackage.loads(task_package)
+        for task_key in group_list:
+            task = signing.SignedPackage.loads(broker.cache.get(task_key))
             if not task['success']:
                 failure_count += 1
         return failure_count
@@ -352,7 +350,10 @@ def delete_group_cached(group_id, broker=None):
     """
     if not broker:
         broker = get_broker()
-    return delete_cached(group_id, broker)
+    group_key = '{}:{}:keys'.format(broker.list_key, group_id)
+    group_list = broker.cache.get(group_key)
+    broker.cache.delete_many(group_list)
+    broker.cache.delete(group_key)
 
 
 def delete_cached(task_id, broker=None):
@@ -361,14 +362,13 @@ def delete_cached(task_id, broker=None):
     """
     if not broker:
         broker = get_broker()
-    key = 'django_q:{}:results'.format(broker.list_key)
-    return broker.cache.delete('{}:{}'.format(key, task_id))
+    return broker.cache.delete('{}:{}'.format(broker.list_key, task_id))
 
 
 def queue_size(broker=None):
     """
     Returns the current queue size.
-    Note that this doesn't count any tasks currently being processed by workers.
+    Note that this doesn't count any tasks curren    key = 'django_q:{}:results'.format(broker.list_key)tly being processed by workers.
 
     :param broker: optional broker
     :return: current queue size
@@ -379,6 +379,26 @@ def queue_size(broker=None):
     return broker.queue_size()
 
 
+def async_iter(func, args_iter, **kwargs):
+    iter_count = len(args_iter)
+    iter_group = uuid()[1]
+    # clean up the kwargs
+    options = kwargs.get('q_options', kwargs)
+    options.pop('hook', None)
+    options['broker'] = options.get('broker', get_broker())
+    options['group'] = iter_group
+    options['iter_count'] = iter_count
+    options['cached'] = True
+    # save the original arguments
+    broker = options['broker']
+    broker.cache.set('{}:{}:args'.format(broker.list_key, iter_group), signing.SignedPackage.dumps(args_iter))
+    for args in args_iter:
+        if type(args) is not tuple:
+            args = (args,)
+        async(func, *args, **options)
+    return iter_group
+
+
 def _sync(pack):
     """Simulate a package travelling through the cluster."""
     task_queue = Queue()

From 0c2da87e2716d30eb2e507755aa1e59c56bc8a17 Mon Sep 17 00:00:00 2001
From: Ilan Steemers <koed00@gmail.com>
Date: Sun, 4 Oct 2015 11:49:07 +0200
Subject: [PATCH 11/14] adds async iter tests

---
 django_q/tests/test_cached.py | 23 ++++++++++++++++++++++-
 1 file changed, 22 insertions(+), 1 deletion(-)

diff --git a/django_q/tests/test_cached.py b/django_q/tests/test_cached.py
index 7e8e618..56380e1 100644
--- a/django_q/tests/test_cached.py
+++ b/django_q/tests/test_cached.py
@@ -4,7 +4,8 @@ import pytest
 
 from django_q.cluster import Sentinel
 from django_q.conf import Conf
-from django_q.tasks import async, result, fetch, count_group, result_group, fetch_group, delete_group, delete_cached
+from django_q.tasks import async, result, fetch, count_group, result_group, fetch_group, delete_group, delete_cached, \
+    async_iter
 from django_q.brokers import get_broker
 
 
@@ -63,3 +64,23 @@ def test_cached(broker):
     assert result(task_id, cached=True) is None
     assert fetch(task_id, cached=True) is None
     broker.cache.clear()
+
+
+@pytest.mark.django_db
+def test_iter(broker):
+    broker.purge_queue()
+    broker.cache.clear()
+    it = [i for i in range(10)]
+    it2 = [(1, -1), (2, -1), (3, -4), (5, 6)]
+    it3 = (1, 2, 3, 4, 5)
+    t = async_iter('math.floor', it, sync=True)
+    t2 = async_iter('math.copysign', it2, sync=True)
+    t3 = async_iter('math.floor', it3, sync=True)
+    t4 = async_iter('math.floor', (1,), sync=True)
+    result_t = result(t)
+    assert result_t is not None
+    task_t = fetch(t)
+    assert task_t.result == result_t
+    assert result(t2) is not None
+    assert result(t3) is not None
+    assert result(t4)[0] == 1

From edf188e7bc83cc338a76b67c9cad8b75e0ad441d Mon Sep 17 00:00:00 2001
From: Ilan Steemers <koed00@gmail.com>
Date: Sun, 4 Oct 2015 11:52:43 +0200
Subject: [PATCH 12/14] adds async iter tests

---
 django_q/tests/test_cached.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/django_q/tests/test_cached.py b/django_q/tests/test_cached.py
index 56380e1..ee05e36 100644
--- a/django_q/tests/test_cached.py
+++ b/django_q/tests/test_cached.py
@@ -80,6 +80,7 @@ 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

From 486a0021c743111aa2c68a4a1d4c813dd4a5c50b Mon Sep 17 00:00:00 2001
From: Ilan Steemers <koed00@gmail.com>
Date: Sun, 4 Oct 2015 12:44:59 +0200
Subject: [PATCH 13/14] docs: adds async_iter

---
 docs/tasks.rst | 47 ++++++++++++++++++++++++++++++++++++++++++++---
 1 file changed, 44 insertions(+), 3 deletions(-)

diff --git a/docs/tasks.rst b/docs/tasks.rst
index 1d0efcd..f1a0fb1 100644
--- a/docs/tasks.rst
+++ b/docs/tasks.rst
@@ -89,6 +89,28 @@ Please not that this will override any other option keywords.
     For tasks to be processed you will need to have a worker cluster running in the background using ``python manage.py qcluster``
     or you need to configure Django Q to run in synchronous mode for testing using the :ref:`sync` option.
 
+
+
+Async 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
+    from django_q.tasks import async_iter, result
+
+    # set up a list of arguments for math.floor
+    iter = [i for i in range(100)]
+
+    # async iter them
+    id=async_iter('math.floor',iter)
+
+    # wait for the collated result for 1 second
+    result_list = result(id, wait=1000)
+
+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.
+Needs the Django cache framework.
+
 .. _groups:
 
 Groups
@@ -103,14 +125,16 @@ You can group together results by passing :func:`async` the optional ``group`` k
     for i in range(4):
         async('math.modf', i, group='modf')
 
-    # after the tasks have finished you can get the group results
-    result = result_group('modf')
+    # wait until the group has 4 results
+    result = result_group('modf', count=4)
     print(result)
 
 .. code-block:: python
 
     [(0.0, 0.0), (0.0, 1.0), (0.0, 2.0), (0.0, 3.0)]
 
+Note that the same can be achieved much faster with :func:`async_iter`
+
 Take care to not limit your results database too much and call :func:`delete_group` before each run, unless you want your results to keep adding up.
 Instead of :func:`result_group` you can also use :func:`fetch_group` to return a queryset of :class:`Task` objects.:
 
@@ -163,7 +187,7 @@ By using a cache backend like Redis or Memcached you can speed up access to your
 
 When you set ``cached=True``, results will be saved permanently in the cache and you will have to rely on your backend's cleanup strategies (like LRU) to
 manage stale results.
-You can however opt to set a manual timeout on the results, by setting ``cached=60``. Meaning the result will be evicted from the cache after 60 seconds.
+You can also opt to set a manual timeout on the results, by setting ``cached=60``. Meaning the result will be evicted from the cache after 60 seconds.
 This works both globally or on individual async executions.::
 
     # simple cached example
@@ -175,6 +199,12 @@ This works both globally or on individual async executions.::
     # wait max 50ms for the result to appear in the cache
     result(id, wait=50, cached=True)
 
+    # o fetch the task object
+    task = fetch(id, cache=True)
+
+    # and then save it to the database
+    task.save()
+
 This also works for group actions::
 
     # cached group example
@@ -195,6 +225,7 @@ This also works for group actions::
     # wait max 50ms for one hundred results to return
     result_group('frexp', wait=50, count=100, cached=True)
 
+Note that exact same result can be achieved by using the more convenient :func:`async_iter` in this case, but without hook support.
 
 Synchronous testing
 -------------------
@@ -287,6 +318,16 @@ Reference
 
     Renamed from get_task
 
+.. py:function:: async_iter(func, args_iter,**kwargs)
+
+   Runs iterable arguments against the cache backend and returns a single collated result
+
+   :param object func: The task function to execute
+   :param args: An iterable containing arguments for the task function
+   :param dict kwargs: Keyword arguments for the task function. Ignores ``cached`` and ``hook``.
+   :returns: The uuid of the task
+   :rtype: str
+
 .. py:function:: queue_size()
 
     Returns the size of the broker queue.

From 5d6efbb3bb92b78a3d12e903c4bede8444f8dc0c Mon Sep 17 00:00:00 2001
From: Ilan Steemers <koed00@gmail.com>
Date: Sun, 4 Oct 2015 14:36:42 +0200
Subject: [PATCH 14/14] docs: updated group example

---
 docs/examples.rst | 38 +++++++++++++++++++++++++++-----------
 1 file changed, 27 insertions(+), 11 deletions(-)

diff --git a/docs/examples.rst b/docs/examples.rst
index 3ab2219..15ce2a2 100644
--- a/docs/examples.rst
+++ b/docs/examples.rst
@@ -253,8 +253,7 @@ Adapted from `Sebastian Raschka's blog <http://sebastianraschka.com/Articles/201
     # Group example with Parzen-window estimation
     import numpy
 
-    from django_q.tasks import async, result_group,\
-        count_group, delete_group
+    from django_q.tasks import async, result_group, delete_group
 
     # the estimation function
     def parzen_estimation(x_samples, point_x, h):
@@ -268,30 +267,47 @@ Adapted from `Sebastian Raschka's blog <http://sebastianraschka.com/Articles/201
                 k_n += 1
         return h, (k_n / len(x_samples)) / (h ** point_x.shape[1])
 
-
     # create 100 calculations and send them to the cluster
     def parzen_async():
         # clear the previous results
-        delete_group('parzen', tasks=True)
+        delete_group('parzen', cached=True)
         mu_vec = numpy.array([0, 0])
         cov_mat = numpy.array([[1, 0], [0, 1]])
-        sample = numpy.random.\
+        sample = numpy.random. \
             multivariate_normal(mu_vec, cov_mat, 10000)
         widths = numpy.linspace(1.0, 1.2, 100)
         x = numpy.array([[0], [0]])
-        # async them with a group label and a hook
+        # async them with a group label to the cache backend
         for w in widths:
             async(parzen_estimation, sample, x, w,
-             group='parzen', hook=parzen_hook)
+                  group='parzen', cached=True)
+        # return after 100 results
+        return result_group('parzen', count=100, cached=True)
 
-    # wait for 100 results to return and print it.
-    def parzen_hook(task):
-        if task.group_count() == 100:
-            print(task.group_result())
 
 
 Django Q is not optimized for distributed computing, but this example will give you an idea of what you can do with task :ref:`groups`.
 
+Alternatively the ``parzen_async()`` function can also be written with :func:`async_iter`, which automatically utilizes the cache backend and groups to return a single result from an iterable:
+
+.. code-block:: python
+
+    # create 100 calculations and send them to the cluster
+    # with async_iter
+    def parzen_async():
+        mu_vec = numpy.array([0, 0])
+        cov_mat = numpy.array([[1, 0], [0, 1]])
+        sample = numpy.random. \
+            multivariate_normal(mu_vec, cov_mat, 10000)
+        widths = numpy.linspace(1.0, 1.2, 100)
+        x = numpy.array([[0], [0]])
+        # async them with async iterable
+        args = [(sample, x, w) for w in widths]
+        result_id = async_iter(parzen_estimation, args)
+        # return the result or timeout after 10 seconds
+        return result(result_id, wait=10000)
+
+
 .. note::
 
     If you have an example you want to share, please submit a pull request on `github <https://github.com/Koed00/django-q/>`__.