Adds an Iter class

The Iter class serves as a convenience rwrapper around the `async_iter` function
This commit is contained in:
Ilan Steemers
2015-10-18 13:53:32 +02:00
parent 083715f9bd
commit eb72eb33d9
2 changed files with 77 additions and 2 deletions

View File

@@ -427,6 +427,67 @@ def async_chain(chain, group=None, cached=Conf.CACHED, sync=Conf.SYNC, broker=No
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
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

View File

@@ -5,7 +5,7 @@ import pytest
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, Chain, async_chain
async_iter, Chain, async_chain, Iter
from django_q.brokers import get_broker
@@ -100,7 +100,21 @@ 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
# 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