From c127604f1054554310490e113d206a356d247ba0 Mon Sep 17 00:00:00 2001 From: Ilan Date: Sat, 18 Jul 2015 17:20:50 +0200 Subject: [PATCH] Adds task groups * task now takes an optional keyword `group` * added result_group() and fetch_group() * added group column to Success admin * group column in searchable * added `name` field to Schedules * Scheduled tasks take schedule name as group --- django_q/__init__.py | 2 +- django_q/admin.py | 8 +++-- django_q/cluster.py | 11 +++--- .../migrations/0005_auto_20150718_1506.py | 24 +++++++++++++ django_q/models.py | 13 ++++++- django_q/tasks.py | 36 ++++++++++++++++--- django_q/tests/test_cluster.py | 8 +++-- 7 files changed, 85 insertions(+), 17 deletions(-) create mode 100644 django_q/migrations/0005_auto_20150718_1506.py diff --git a/django_q/__init__.py b/django_q/__init__.py index 7e5a656..88010c5 100644 --- a/django_q/__init__.py +++ b/django_q/__init__.py @@ -4,7 +4,7 @@ import sys myPath = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, myPath) -from .tasks import async, schedule, result, fetch +from .tasks import async, schedule, result, result_group, fetch, fetch_group from .models import Task, Schedule, Success, Failure from .cluster import Cluster diff --git a/django_q/admin.py b/django_q/admin.py index 7dacac5..2f80ade 100644 --- a/django_q/admin.py +++ b/django_q/admin.py @@ -11,7 +11,8 @@ class TaskAdmin(admin.ModelAdmin): 'func', 'started', 'stopped', - 'time_taken' + 'time_taken', + 'group' ) def has_add_permission(self, request, obj=None): @@ -23,7 +24,7 @@ class TaskAdmin(admin.ModelAdmin): qs = super(TaskAdmin, self).get_queryset(request) return qs.filter(success=True) - search_fields = ('name', 'func') + search_fields = ('name', 'func', 'group') readonly_fields = [] def get_readonly_fields(self, request, obj=None): @@ -65,6 +66,7 @@ class FailAdmin(admin.ModelAdmin): class ScheduleAdmin(admin.ModelAdmin): list_display = ( 'id', + 'name', 'func', 'schedule_type', 'repeats', @@ -75,7 +77,7 @@ class ScheduleAdmin(admin.ModelAdmin): list_filter = ('next_run', 'schedule_type') search_fields = ('func',) - list_display_links = ('id', 'func') + list_display_links = ('id', 'name') admin.site.register(Schedule, ScheduleAdmin) diff --git a/django_q/cluster.py b/django_q/cluster.py index ad2edab..6ad88c2 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -404,12 +404,13 @@ def save_task(task): Task.objects.create(id=task['id'], name=task['name'], func=task['func'], - hook=task['hook'], + hook=task.get('hook'), args=task['args'], kwargs=task['kwargs'], started=task['started'], stopped=task['stopped'], result=task['result'], + group=task.get('group'), success=task['success']) except Exception as e: logger.error(e) @@ -455,13 +456,15 @@ def scheduler(list_key=Conf.Q_LIST): s.repeats += -1 # send it to the cluster kwargs['list_key'] = list_key + kwargs['group'] = s.name or s.id s.task = tasks.async(s.func, *args, **kwargs) # log it if not s.task: - logger.error(_('{} failed to create a task from schedule {} [{}]').format(current_process().name, s.id), - s.func) + logger.error( + _('{} failed to create a task from schedule [{}]').format(current_process().name, s.name or s.id)) else: - logger.info(_('{} created a task from schedule {} [{}]').format(current_process().name, s.id, s.func)) + logger.info( + _('{} created a task from schedule [{}]').format(current_process().name, s.name or s.id)) # default behavior is to delete a ONCE schedule if s.schedule_type == s.ONCE: if s.repeats < 0: diff --git a/django_q/migrations/0005_auto_20150718_1506.py b/django_q/migrations/0005_auto_20150718_1506.py new file mode 100644 index 0000000..e0636c6 --- /dev/null +++ b/django_q/migrations/0005_auto_20150718_1506.py @@ -0,0 +1,24 @@ +# -*- coding: utf-8 -*- +from __future__ import unicode_literals + +from django.db import models, migrations + + +class Migration(migrations.Migration): + + dependencies = [ + ('django_q', '0004_auto_20150710_1043'), + ] + + operations = [ + migrations.AddField( + model_name='schedule', + name='name', + field=models.CharField(max_length=256, null=True), + ), + migrations.AddField( + model_name='task', + name='group', + field=models.CharField(max_length=100, null=True, editable=False), + ), + ] diff --git a/django_q/models.py b/django_q/models.py index 98b3b62..21e6ec1 100644 --- a/django_q/models.py +++ b/django_q/models.py @@ -1,6 +1,6 @@ -import importlib import logging +import importlib from django.core.urlresolvers import reverse from django.utils.translation import ugettext_lazy as _ from django.db import models @@ -18,6 +18,7 @@ class Task(models.Model): args = PickledObjectField(null=True) kwargs = PickledObjectField(null=True) result = PickledObjectField(null=True) + group = models.CharField(max_length=100, editable=False, null=True) started = models.DateTimeField(editable=False) stopped = models.DateTimeField(editable=False) success = models.BooleanField(default=True, editable=False) @@ -29,6 +30,11 @@ class Task(models.Model): elif Task.objects.filter(name=task_id).exists(): return Task.objects.get(name=task_id).result + @staticmethod + def get_group_result(group_id): + # values_list() doesn't work here cause it returns encoded fields + return [t.result for t in Task.get_task_group(group_id)] + @staticmethod def get_task(task_id): if len(task_id) == 32 and Task.objects.filter(id=task_id).exists(): @@ -36,6 +42,10 @@ class Task(models.Model): elif Task.objects.filter(name=task_id).exists(): return Task.objects.get(name=task_id) + @staticmethod + def get_task_group(group_id): + return Task.objects.filter(group=group_id) + def time_taken(self): return (self.stopped - self.started).total_seconds() @@ -101,6 +111,7 @@ class Failure(Task): class Schedule(models.Model): + name = models.CharField(max_length=256, null=True) func = models.CharField(max_length=256, help_text='e.g. module.tasks.function') hook = models.CharField(max_length=256, null=True, blank=True, help_text='e.g. module.tasks.result_function') args = models.TextField(null=True, blank=True, help_text=_("e.g. 1, 2, 'John'")) diff --git a/django_q/tasks.py b/django_q/tasks.py index 9747fd1..9cc3ef6 100644 --- a/django_q/tasks.py +++ b/django_q/tasks.py @@ -25,20 +25,27 @@ def async(func, *args, **kwargs): # optional list_key list_key = kwargs.pop('list_key', Conf.Q_LIST) # optional redis connection - r = kwargs.pop('redis', redis_client) + redis = kwargs.pop('redis', redis_client) # optional sync mode - s = kwargs.pop('sync', False) + sync = kwargs.pop('sync', False) + # optional group + group = kwargs.pop('group', None) # get an id tag = uuid() # build the task package - task = {'id': tag[1], 'name': tag[0], 'func': func, 'hook': hook, 'args': args, 'kwargs': kwargs, + task = {'id': tag[1], 'name': tag[0], 'func': func, 'args': args, 'kwargs': kwargs, 'started': timezone.now()} + # add optionals + if hook: + task['hook'] = hook + if group: + task['group'] = group # sign it pack = signing.SignedPackage.dumps(task) - if s: + if sync: return _sync(task['id'], pack) # push it - r.rpush(list_key, pack) + redis.rpush(list_key, pack) logger.debug('Pushed {}'.format(tag)) return task['id'] @@ -83,6 +90,15 @@ def result(task_id): return Task.get_result(task_id) +def result_group(group_id): + """ + returns a list of results for a task group + :param str group_id: the group id + :return: list or results + """ + return Task.get_group_result(group_id) + + def fetch(task_id): """ Returns the processed task @@ -94,6 +110,16 @@ def fetch(task_id): return Task.get_task(task_id) +def fetch_group(group_id): + """ + Returns a list of Tasks for a task group + :param str group_id: the group id + :return: list of Tasks + """ + + return Task.get_task_group(group_id) + + def _sync(task_id, pack): """ Simulates a package travelling through the cluster. diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py index 32fba17..c51ea08 100644 --- a/django_q/tests/test_cluster.py +++ b/django_q/tests/test_cluster.py @@ -10,7 +10,7 @@ sys.path.insert(0, myPath + '/../') from django_q.cluster import Cluster, Sentinel, pusher, worker, monitor from django_q.humanhash import DEFAULT_WORDLIST -from django_q.tasks import fetch, async, result +from django_q.tasks import fetch, fetch_group, async, result, result_group from django_q.models import Task from django_q.conf import Conf, redis_client from .tasks import multiply @@ -119,10 +119,10 @@ def test_async(r, admin_user): f = async(multiply, 753, 2, hook=assert_result, list_key=list_key, redis=r) # model as argument g = async('django_q.tests.tasks.get_task_name', Task(name='John'), list_key=list_key, redis=r) - # args and kwargs and broken hook + # args,kwargs, group and broken hook h = async('django_q.tests.tasks.word_multiply', 2, word='django', hook='fail.me', list_key=list_key, redis=r) # args unpickle test - j = async('django_q.tests.tasks.get_user_id', admin_user, list_key=list_key, redis=r) + j = async('django_q.tests.tasks.get_user_id', admin_user, list_key=list_key, group='test_j', redis=r) # check if everything has a task id assert isinstance(a, str) assert isinstance(b, str) @@ -197,6 +197,8 @@ def test_async(r, admin_user): assert result_j is not None assert result_j.success is True assert result_j.result == result_j.args[0].id + assert result_group('test_j') == [result_j.result] + assert fetch_group('test_j')[0].id == [result_j][0].id r.delete(list_key)