Merge pull request #87 from Koed00/dev

adds `cached` option to `async_iter`

django_q/cluster.py

@@ -420,7 +420,7 @@ def save_cached(task, broker):
         timeout = None
     try:
         group = task.get('group', False)
-        iter_count = task.get('iter_count', 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'])
@@ -435,7 +435,13 @@ def save_cached(task, broker):
                 task['result'] = results
                 task['id'] = group
                 task['args'] = signing.SignedPackage.loads(broker.cache.get(group_args))
-                save_task(task)
+                task.pop('iter_count', None)
+                task.pop('group', None)
+                if task.get('iter_cached', None):
+                    task['cached'] = task.pop('iter_cached', None)
+                    save_cached(task, broker=broker)
+                else:
+                    save_task(task)
                 broker.cache.delete_many(group_list)
                 broker.cache.delete_many([group_key, group_args])
                 return

django_q/tasks.py

@@ -25,6 +25,7 @@ def async(func, *args, **kwargs):
     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
     tag = uuid()
     # build the task package
@@ -44,6 +45,8 @@ def async(func, *args, **kwargs):
         task['cached'] = cached
     if iter_count:
         task['iter_count'] = iter_count
+    if iter_cached:
+        task['iter_cached'] = iter_cached
     # sign it
     pack = signing.SignedPackage.dumps(task)
     if sync or Conf.SYNC:
@@ -388,6 +391,8 @@ def async_iter(func, args_iter, **kwargs):
     options['broker'] = options.get('broker', get_broker())
     options['group'] = iter_group
     options['iter_count'] = iter_count
+    if options.get('cached', None):
+        options['iter_cached'] = options['cached']
     options['cached'] = True
     # save the original arguments
     broker = options['broker']

django_q/tests/test_cached.py

@@ -1,8 +1,8 @@
-from multiprocessing import Event
+from multiprocessing import Event, Queue, Value
 
 import pytest
+from django_q.cluster import pusher, worker, monitor
 
-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, \
     async_iter
@@ -33,6 +33,7 @@ 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.popysign', 1, -1, cached=True, broker=broker, group=group)
+    iter_id = async_iter('math.floor', [i for i in range(10)], cached=True)
     # test wait on cache
     # test wait timeout
     assert result(task_id, wait=10, cached=True) is None
@@ -41,11 +42,23 @@ def test_cached(broker):
     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()
+    # run a single inline cluster
+    task_count = 17
+    assert broker.queue_size() == task_count
+    task_queue = Queue()
     stop_event = Event()
     stop_event.set()
-    Sentinel(stop_event, start_event, broker=broker)
+    for i in range(task_count):
+        pusher(task_queue, stop_event, broker=broker)
+    assert broker.queue_size() == 0
+    assert task_queue.qsize() == task_count
+    task_queue.put('STOP')
+    result_queue = Queue()
+    worker(task_queue, result_queue, Value('f', -1))
+    assert result_queue.qsize() == task_count
+    result_queue.put('STOP')
+    monitor(result_queue)
+    assert result_queue.qsize() == 0
     # assert results
     assert result(task_id, wait=500, cached=True) == -1
     assert fetch(task_id, wait=500, cached=True).result == -1
@@ -63,6 +76,9 @@ def test_cached(broker):
     delete_cached(task_id)
     assert result(task_id, cached=True) is None
     assert fetch(task_id, cached=True) is None
+    # iter cached
+    assert result(iter_id) is None
+    assert result(iter_id, cached=True) is not None
     broker.cache.clear()
 
 
@@ -85,3 +101,4 @@ def test_iter(broker):
     assert result(t2) is not None
     assert result(t3) is not None
     assert result(t4)[0] == 1
+    # test cached iter result

docs/tasks.rst

@@ -320,11 +320,12 @@ Reference
 
 .. py:function:: async_iter(func, args_iter,**kwargs)
 
-   Runs iterable arguments against the cache backend and returns a single collated result
+   Runs iterable arguments against the cache backend and returns a single collated result.
+   Accepts the same options as :func:`async` except ``hook``.
 
    :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``.
+   :param dict kwargs: Keyword arguments for the task function. Ignores ``hook``.
    :returns: The uuid of the task
    :rtype: str