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

@@ -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
 

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)

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:

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),
+        ),
+    ]

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'"))

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.

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)