diff --git a/django_q/tasks.py b/django_q/tasks.py index b36a9ba..70d1539 100644 --- a/django_q/tasks.py +++ b/django_q/tasks.py @@ -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 diff --git a/django_q/tests/test_cached.py b/django_q/tests/test_cached.py index 58bf11b..00bfe19 100644 --- a/django_q/tests/test_cached.py +++ b/django_q/tests/test_cached.py @@ -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