diff --git a/README.rst b/README.rst index 44679ef..c4f60cf 100644 --- a/README.rst +++ b/README.rst @@ -127,7 +127,10 @@ Use `async` from your code to quickly offload tasks: task_result = result(task_id) # result returns None if the task has not been executed yet - # so in most cases you will want to use a hook: + # you can wait for it + task_result = result(task_id, 200) + + # but in most cases you will want to use a hook: async('math.modf', 2.5, hook='hooks.print_result') diff --git a/django_q/tasks.py b/django_q/tasks.py index 17a8952..62da8f1 100644 --- a/django_q/tasks.py +++ b/django_q/tasks.py @@ -5,6 +5,7 @@ from multiprocessing import Queue, Value from django.utils import timezone # local +import time import signing import cluster from django_q.conf import Conf, logger @@ -82,16 +83,25 @@ def schedule(func, *args, **kwargs): ) -def result(task_id): +def result(task_id, wait=0): """ Return the result of the named task. :type task_id: str or uuid :param task_id: the task name or uuid + :type wait: int + :param wait: number of milliseconds to wait for a result :return: the result object of this task :rtype: object """ - return Task.get_result(task_id) + start = time.time() + while True: + r = Task.get_result(task_id) + if r: + return r + if (time.time() - start) * 1000 >= wait: + break + time.sleep(0.01) def result_group(group_id, failures=False): @@ -105,16 +115,25 @@ def result_group(group_id, failures=False): return Task.get_result_group(group_id, failures) -def fetch(task_id): +def fetch(task_id, wait=0): """ Return the processed task. :param task_id: the task name or uuid :type task_id: str or uuid + :param wait: the number of milliseconds to wait for a result + :type wait: int :return: the full task object :rtype: Task """ - return Task.get_task(task_id) + start = time.time() + while True: + t = Task.get_task(task_id) + if t: + return t + if (time.time() - start) * 1000 >= wait: + break + time.sleep(0.01) def fetch_group(group_id, failures=True): diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py index bc21f97..5f9c927 100644 --- a/django_q/tests/test_cluster.py +++ b/django_q/tests/test_cluster.py @@ -230,6 +230,8 @@ def test_async(broker, admin_user): assert result_j.group_delete(tasks=True) is None # task k should not have been saved assert fetch(k) is None + assert fetch(k, 100) is None + assert result(k, 100) is None broker.delete_queue() diff --git a/docs/tasks.rst b/docs/tasks.rst index 178cc77..ac88c47 100644 --- a/docs/tasks.rst +++ b/docs/tasks.rst @@ -25,7 +25,10 @@ Use :func:`async` from your code to quickly offload tasks to the :class:`Cluster task_result = result(task_id) # result returns None if the task has not been executed yet - # so in most cases you will want to use a hook: + # you can wait for it + task_result = result(task_id, 200) + + # but in most cases you will want to use a hook: async('math.modf', 2.5, hook='hooks.print_result') @@ -208,19 +211,21 @@ Reference :returns: The uuid of the task :rtype: str -.. py:function:: result(task_id) +.. py:function:: result(task_id, wait=0) 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 :returns: The result of the executed task -.. py:function:: fetch(task_id) +.. py:function:: fetch(task_id, wait=0) Returns a previously executed task :param str name: the uuid or name of the task - :returns: The task if any + :param in wait: optional milliseconds to wait for a result + :returns: A task object :rtype: Task .. versionchanged:: 0.2.0