Adds wait option to result() and fetch()

This commit is contained in:
Ilan Steemers
2015-09-18 10:57:09 +02:00
parent a30e5d2c28
commit a493f9cc78
4 changed files with 38 additions and 9 deletions
+4 -1
View File
@@ -127,7 +127,10 @@ Use `async` from your code to quickly offload tasks:
task_result = result(task_id) task_result = result(task_id)
# result returns None if the task has not been executed yet # 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') async('math.modf', 2.5, hook='hooks.print_result')
+23 -4
View File
@@ -5,6 +5,7 @@ from multiprocessing import Queue, Value
from django.utils import timezone from django.utils import timezone
# local # local
import time
import signing import signing
import cluster import cluster
from django_q.conf import Conf, logger 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. Return the result of the named task.
:type task_id: str or uuid :type task_id: str or uuid
:param task_id: the task name 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 :return: the result object of this task
:rtype: object :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): 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) return Task.get_result_group(group_id, failures)
def fetch(task_id): def fetch(task_id, wait=0):
""" """
Return the processed task. Return the processed task.
:param task_id: the task name or uuid :param task_id: the task name or uuid
:type task_id: str 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 :return: the full task object
:rtype: Task :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): def fetch_group(group_id, failures=True):
+2
View File
@@ -230,6 +230,8 @@ def test_async(broker, admin_user):
assert result_j.group_delete(tasks=True) is None assert result_j.group_delete(tasks=True) is None
# task k should not have been saved # task k should not have been saved
assert fetch(k) is None assert fetch(k) is None
assert fetch(k, 100) is None
assert result(k, 100) is None
broker.delete_queue() broker.delete_queue()
+9 -4
View File
@@ -25,7 +25,10 @@ Use :func:`async` from your code to quickly offload tasks to the :class:`Cluster
task_result = result(task_id) task_result = result(task_id)
# result returns None if the task has not been executed yet # 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') async('math.modf', 2.5, hook='hooks.print_result')
@@ -208,19 +211,21 @@ Reference
:returns: The uuid of the task :returns: The uuid of the task
:rtype: str :rtype: str
.. py:function:: result(task_id) .. py:function:: result(task_id, wait=0)
Gets the result of a previously executed task Gets the result of a previously executed task
:param str task_id: the uuid or name of the 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 :returns: The result of the executed task
.. py:function:: fetch(task_id) .. py:function:: fetch(task_id, wait=0)
Returns a previously executed task Returns a previously executed task
:param str name: the uuid or name of the 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 :rtype: Task
.. versionchanged:: 0.2.0 .. versionchanged:: 0.2.0