fork/django-q2
1
0
Fork 0
mirror of https://github.com/django-q2/django-q2.git synced 2026-10-05 21:38:12 +08:00
Code Issues Packages Projects Releases Wiki Activity

adds count and wait option to group result and fetch

Browse Source 
result_group and fetch_group can now be told to block for a certain number of results or a number of miliseconds or a combination of both.
This commit is contained in:
Ilan Steemers
2015-10-02 18:29:03 +02:00
parent 9cfc41f1cb
commit 81d5abbd00
1 changed files with 87 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
django_q/tasks.py
+87 -42
View File
@@ -15,7 +15,7 @@ from django_q.brokers import get_broker
def async(func, *args, **kwargs): def async(func, *args, **kwargs):
"""Send a task to the cluster.""" """Queue a task for the cluster."""
# get options from q_options dict or direct from kwargs # get options from q_options dict or direct from kwargs
options = kwargs.pop('q_options', kwargs) options = kwargs.pop('q_options', kwargs)
hook = options.pop('hook', None) hook = options.pop('hook', None)
@@ -94,7 +94,7 @@ def result(task_id, wait=0, cached=Conf.CACHED):
:param task_id: the task name or uuid :param task_id: the task name or uuid
:type wait: int :type wait: int
:param wait: number of milliseconds to wait for a result :param wait: number of milliseconds to wait for a result
:param cached: run this against the cache backend :param bool cached: run this against the cache backend
:return: the result object of this task :return: the result object of this task
:rtype: object :rtype: object
""" """
@@ -127,35 +127,58 @@ def result_cached(task_id, wait=0, broker=None):
time.sleep(0.01) time.sleep(0.01)
def result_group(group_id, failures=False, cached=Conf.CACHED): def result_group(group_id, failures=False, wait=0, count=None, cached=Conf.CACHED):
""" """
Return a list of results for a task group. Return a list of results for a task group.
:param str group_id: the group id :param str group_id: the group id
:param bool failures: set to True to include failures :param bool failures: set to True to include failures
:param cached: run this against the cache backend :param int count: Block until there are this many results in the group
:param bool cached: run this against the cache backend
:return: list or results :return: list or results
""" """
if cached: if cached:
return result_group_cached(group_id, failures) return result_group_cached(group_id, failures, wait, count)
return Task.get_result_group(group_id, failures) start = time.time()
if count:
while True:
if count_group(group_id) == count or wait and (time.time() - start) * 1000 >= wait:
break
time.sleep(0.01)
while True:
r = Task.get_result_group(group_id, failures)
if r:
return r
if (time.time() - start) * 1000 >= wait:
break
time.sleep(0.01)
def result_group_cached(group_id, failures=False, broker=None): def result_group_cached(group_id, failures=False, wait=0, count=None, broker=None):
""" """
Return a list of results for a task group from the cache backend Return a list of results for a task group from the cache backend
""" """
if not broker: if not broker:
broker = get_broker() broker = get_broker()
start = time.time()
if count:
while True:
if count_group_cached(group_id) == count or wait and (time.time() - start) * 1000 >= wait:
break
time.sleep(0.01)
key = 'django_q:{}:results'.format(broker.list_key) key = 'django_q:{}:results'.format(broker.list_key)
group_list = broker.cache.get('{}:{}'.format(key, group_id)) while True:
if group_list: group_list = broker.cache.get('{}:{}'.format(key, group_id))
result_list = [] if group_list:
for task_package in group_list: result_list = []
task = signing.SignedPackage.loads(task_package) for task_package in group_list:
if task['success'] or failures: task = signing.SignedPackage.loads(task_package)
result_list.append(task['result']) if task['success'] or failures:
return result_list result_list.append(task['result'])
return result_list
if (time.time() - start) * 1000 >= wait:
break
time.sleep(0.01)
def fetch(task_id, wait=0, cached=Conf.CACHED): def fetch(task_id, wait=0, cached=Conf.CACHED):
@@ -166,7 +189,7 @@ def fetch(task_id, wait=0, cached=Conf.CACHED):
:type task_id: str or uuid :type task_id: str or uuid
:param wait: the number of milliseconds to wait for a result :param wait: the number of milliseconds to wait for a result
:type wait: int :type wait: int
:param cached: run this against the cache backend :param bool cached: run this against the cache backend
:return: the full task object :return: the full task object
:rtype: Task :rtype: Task
""" """
@@ -210,46 +233,68 @@ def fetch_cached(task_id, wait=0, broker=None):
time.sleep(0.01) time.sleep(0.01)
def fetch_group(group_id, failures=True, cached=Conf.CACHED): def fetch_group(group_id, failures=True, wait=0, count=None, cached=Conf.CACHED):
""" """
Return a list of Tasks for a task group. Return a list of Tasks for a task group.
:param str group_id: the group id :param str group_id: the group id
:param bool failures: set to False to exclude failures :param bool failures: set to False to exclude failures
:param cached: run this against the cache backend :param bool cached: run this against the cache backend
:return: list of Tasks :return: list of Tasks
""" """
if cached: if cached:
return fetch_group_cached(group_id, failures) return fetch_group_cached(group_id, failures, wait, count)
return Task.get_task_group(group_id, failures) start = time.time()
if count:
while True:
if count_group(group_id) == count or wait and (time.time() - start) * 1000 >= wait:
break
time.sleep(0.01)
while True:
r = Task.get_task_group(group_id, failures)
if r:
return r
if (time.time() - start) * 1000 >= wait:
break
time.sleep(0.01)
def fetch_group_cached(group_id, failures=True, broker=None): def fetch_group_cached(group_id, failures=True, wait=0, count=None, broker=None):
""" """
Return a list of Tasks for a task group in the cache backend Return a list of Tasks for a task group in the cache backend
""" """
if not broker: if not broker:
broker = get_broker() broker = get_broker()
start = time.time()
if count:
while True:
if count_group_cached(group_id) == count or wait and (time.time() - start) * 1000 >= wait:
break
time.sleep(0.01)
key = 'django_q:{}:results'.format(broker.list_key) key = 'django_q:{}:results'.format(broker.list_key)
group_list = broker.cache.get('{}:{}'.format(key, group_id)) while True:
if group_list: group_list = broker.cache.get('{}:{}'.format(key, group_id))
task_list = [] if group_list:
for task_package in group_list: task_list = []
task = signing.SignedPackage.loads(task_package) for task_package in group_list:
if task['success'] or failures: task = signing.SignedPackage.loads(task_package)
t = Task(id=task['id'], if task['success'] or failures:
name=task['name'], t = Task(id=task['id'],
func=task['func'], name=task['name'],
hook=task.get('hook'), func=task['func'],
args=task['args'], hook=task.get('hook'),
kwargs=task['kwargs'], args=task['args'],
started=task['started'], kwargs=task['kwargs'],
stopped=task['stopped'], started=task['started'],
result=task['result'], stopped=task['stopped'],
group=task.get('group'), result=task['result'],
success=task['success']) group=task.get('group'),
task_list.append(t) success=task['success'])
return task_list task_list.append(t)
return task_list
if (time.time() - start) * 1000 >= wait:
break
time.sleep(0.01)
def count_group(group_id, failures=False, cached=Conf.CACHED): def count_group(group_id, failures=False, cached=Conf.CACHED):
@@ -258,7 +303,7 @@ def count_group(group_id, failures=False, cached=Conf.CACHED):
:param str group_id: the group id :param str group_id: the group id
:param bool failures: Returns failure count if True :param bool failures: Returns failure count if True
:param cached: run this against the cache backend :param bool cached: run this against the cache backend
:return: the number of tasks/results in a group :return: the number of tasks/results in a group
:rtype: int :rtype: int
""" """
@@ -293,7 +338,7 @@ def delete_group(group_id, tasks=False, cached=Conf.CACHED):
:param str group_id: the group id :param str group_id: the group id
:param bool tasks: If set to True this will also delete the group tasks. :param bool tasks: If set to True this will also delete the group tasks.
Otherwise just the group label is removed. Otherwise just the group label is removed.
:param cached: run this against the cache backend :param bool cached: run this against the cache backend
:return: :return:
""" """
if cached: if cached: