Merge pull request #64 from Koed00/dev

Adds configuration output and other enhancements

django_q/admin.py

@@ -98,6 +98,9 @@ class QueueAdmin(admin.ModelAdmin):
     list_display = (
         'id',
         'key',
+        'task_id',
+        'name',
+        'func',
         'lock'
     )
 

django_q/brokers/__init__.py

@@ -7,7 +7,6 @@ class Broker(object):
         self.connection = self.get_connection(list_key)
         self.list_key = list_key
         self.cache = self.get_cache()
-        self.task_cache = []
 
     def enqueue(self, task):
         """

django_q/brokers/aws_sqs.py

@@ -17,17 +17,9 @@ class Sqs(Broker):
         # sqs supports max 10 messages in bulk
         if Conf.BULK > 10:
             Conf.BULK = 10
-        t = None
+        tasks = self.queue.receive_messages(MaxNumberOfMessages=Conf.BULK, VisibilityTimeout=Conf.RETRY)
-        if len(self.task_cache) > 0:
+        if tasks:
-            t = self.task_cache.pop()
+            return [(t.receipt_handle, t.body) for t in tasks]
-        else:
-            tasks = self.queue.receive_messages(MaxNumberOfMessages=Conf.BULK, VisibilityTimeout=Conf.RETRY)
-            if tasks:
-                t = tasks.pop()
-                if tasks:
-                    self.task_cache = tasks
-        if t:
-            return t.receipt_handle, t.body
 
     def acknowledge(self, task_id):
         return self.delete(task_id)

django_q/brokers/disque.py

@@ -11,18 +11,10 @@ class Disque(Broker):
             'ADDJOB {} {} 500 RETRY {}'.format(self.list_key, task, retry)).decode()
 
     def dequeue(self):
-        t = None
-        if len(self.task_cache) > 0:
-            t = self.task_cache.pop()
-        else:
             tasks = self.connection.execute_command(
                 'GETJOB COUNT {} TIMEOUT 1000 FROM {}'.format(Conf.BULK, self.list_key))
             if tasks:
-                t = tasks.pop()
+                return [(t[1].decode(), t[2].decode()) for t in tasks]
-                if tasks:
-                    self.task_cache = tasks
-        if t:
-            return t[1].decode(), t[2].decode()
 
     def queue_size(self):
         return self.connection.execute_command('QLEN {}'.format(self.list_key))

django_q/brokers/ironmq.py

@@ -9,18 +9,10 @@ class IronMQBroker(Broker):
         return self.connection.post(task)['ids'][0]
 
     def dequeue(self):
-        t = None
-        if len(self.task_cache) > 0:
-            t = self.task_cache.pop()
-        else:
             timeout = Conf.RETRY or None
             tasks = self.connection.get(timeout=timeout, wait=1, max=Conf.BULK)['messages']
             if tasks:
-                t = tasks.pop()
+                return [(t['id'], t['body']) for t in tasks]
-                if tasks:
-                    self.task_cache = tasks
-        if t:
-            return t['id'], t['body']
 
     def ping(self):
         return self.connection.name == self.list_key

django_q/brokers/orm.py

@@ -31,24 +31,13 @@ class ORM(Broker):
         return package.pk
 
     def dequeue(self):
-        if len(self.task_cache) > 0:
-            t = self.task_cache.pop()
-            return t.pk, t.payload
-        else:
-            # Get new and timed out tasks
             tasks = OrmQ.objects.using(Conf.ORM).filter(
                 Q(key=self.list_key, lock__isnull=True) |
                 Q(key=self.list_key, lock__lte=timezone.now() - timedelta(seconds=Conf.RETRY)))[:Conf.BULK]
             if tasks:
                 # lock them
                 OrmQ.objects.using(Conf.ORM).filter(pk__in=tasks).update(lock=timezone.now())
-                tasks = [t for t in tasks]
+                return [(t.pk, t.payload) for t in tasks]
-                # pop one task
-                t = tasks.pop()
-                if tasks:
-                    # add remainder to cache
-                    self.task_cache = [t for t in tasks]
-                return t.pk, t.payload
             # empty queue, spare the cpu
             sleep(0.2)
 

django_q/brokers/redis_broker.py

@@ -19,7 +19,7 @@ class Redis(Broker):
     def dequeue(self):
         task = self.connection.blpop(self.list_key, 1)
         if task:
-            return None, task[1]
+            return [(None, task[1])]
 
     def queue_size(self):
         return self.connection.llen(self.list_key)

django_q/cluster.py

@@ -171,7 +171,7 @@ class Sentinel(object):
         self.spawn_process(worker, self.task_queue, self.result_queue, Value('f', -1), self.timeout)
 
     def spawn_monitor(self):
-        return self.spawn_process(monitor, self.result_queue)
+        return self.spawn_process(monitor, self.result_queue, self.broker)
 
     def reincarnate(self, process):
         """
@@ -298,23 +298,24 @@ def pusher(task_queue, event, broker=None):
     logger.info(_('{} pushing tasks at {}').format(current_process().name, current_process().pid))
     while True:
         try:
-            task = broker.dequeue()
+            task_set = broker.dequeue()
         except Exception as e:
             logger.error(e)
             # broker probably crashed. Let the sentinel handle it.
             sleep(10)
             break
-        if task:
+        if task_set:
-            ack_id = task[0]
+            for task in task_set:
-            # unpack the task
+                ack_id = task[0]
-            try:
+                # unpack the task
-                task = signing.SignedPackage.loads(task[1])
+                try:
-            except (TypeError, signing.BadSignature) as e:
+                    task = signing.SignedPackage.loads(task[1])
-                logger.error(e)
+                except (TypeError, signing.BadSignature) as e:
-                broker.fail(ack_id)
+                    logger.error(e)
-                continue
+                    broker.fail(ack_id)
-            task['ack_id'] = ack_id
+                    continue
-            task_queue.put(task)
+                task['ack_id'] = ack_id
+                task_queue.put(task)
             logger.debug(_('queueing from {}').format(broker.list_key))
         if event.is_set():
             break
@@ -493,7 +494,7 @@ def set_cpu_affinity(n, process_ids, actual=not Conf.TESTING):
     """
     Sets the cpu affinity for the supplied processes.
     Requires the optional psutil module.
-    :param int n:
+    :param int n: affinity
     :param list process_ids: a list of pids
     :param bool actual: Test workaround for Travis not supporting cpu affinity
     """

django_q/management/commands/qinfo.py

@@ -1,6 +1,8 @@
+from optparse import make_option
 from django.core.management.base import BaseCommand
 from django.utils.translation import ugettext as _
 
+from django_q.conf import Conf
 from django_q.monitor import info
 
 
@@ -8,5 +10,19 @@ class Command(BaseCommand):
     # Translators: help text for qinfo management command
     help = _('General information over all clusters.')
 
+    option_list = BaseCommand.option_list + (
+        make_option('--config',
+                    action='store_true',
+                    dest='config',
+                    default=False,
+                    help='Print current configuration.'),
+    )
+
     def handle(self, *args, **options):
-        info()
+        if options.get('config', False):
+            hide = ['conf', 'IDLE', 'STOPPING', 'STARTING', 'WORKING', 'SIGNAL_NAMES', 'STOPPED']
+            settings = [a for a in dir(Conf) if not a.startswith('__') and a not in hide]
+            for setting in settings:
+                self.stdout.write('{}: {}'.format(setting, getattr(Conf, setting)))
+        else:
+            info()

django_q/models.py

@@ -1,6 +1,6 @@
 import logging
-from django import get_version
 
+from django import get_version
 import importlib
 from django.core.urlresolvers import reverse
 from django.utils.translation import ugettext_lazy as _
@@ -11,6 +11,8 @@ from django.utils import timezone
 from picklefield import PickledObjectField
 from picklefield.fields import dbsafe_decode
 
+from django_q.signing import SignedPackage
+
 
 class Task(models.Model):
     id = models.CharField(max_length=32, primary_key=True, editable=False)
@@ -40,12 +42,20 @@ class Task(models.Model):
             values = Task.objects.filter(group=group_id).exclude(success=False).values_list('result', flat=True)
         return decode_results(values)
 
+    def group_result(self, failures=False):
+        if self.group:
+            return self.get_result_group(self.group, failures)
+
     @staticmethod
     def get_group_count(group_id, failures=False):
         if failures:
             return Failure.objects.filter(group=group_id).count()
         return Task.objects.filter(group=group_id).count()
 
+    def group_count(self, failures=False):
+        if self.group:
+            return self.get_group_count(self.group, failures)
+
     @staticmethod
     def delete_group(group_id, objects=False):
         group = Task.objects.filter(group=group_id)
@@ -53,6 +63,10 @@ class Task(models.Model):
             return group.delete()
         return group.update(group=None)
 
+    def group_delete(self, tasks=False):
+        if self.group:
+            return self.delete_group(self.group, tasks)
+
     @staticmethod
     def get_task(task_id):
         if len(task_id) == 32 and Task.objects.filter(id=task_id).exists():
@@ -189,14 +203,26 @@ class Schedule(models.Model):
 
 
 class OrmQ(models.Model):
-        key = models.CharField(max_length=100)
+    key = models.CharField(max_length=100)
-        payload = models.TextField()
+    payload = models.TextField()
-        lock = models.DateTimeField(null=True)
+    lock = models.DateTimeField(null=True)
 
-        class Meta:
+    def task(self):
-            app_label = 'django_q'
+        return SignedPackage.loads(self.payload)
-            verbose_name = _('Queued task')
+
-            verbose_name_plural = _('Queued tasks')
+    def func(self):
+        return self.task()['func']
+
+    def task_id(self):
+        return self.task()['id']
+
+    def name(self):
+        return self.task()['name']
+
+    class Meta:
+        app_label = 'django_q'
+        verbose_name = _('Queued task')
+        verbose_name_plural = _('Queued tasks')
 
 
 # Backwards compatibility for Django 1.7
@@ -205,5 +231,3 @@ def decode_results(values):
         # decode values in 1.7
         return [dbsafe_decode(v) for v in values]
     return values
-
-

django_q/tests/test_admin.py

@@ -7,12 +7,13 @@ from django_q.tasks import schedule
 from django_q.models import Task, Failure, OrmQ
 from django_q.humanhash import uuid
 from django_q.conf import Conf
+from django_q.signing import SignedPackage
 
 
 @pytest.mark.django_db
 def test_admin_views(admin_client):
-    Conf.ORM='default'
+    Conf.ORM = 'default'
-    s = schedule('sched.test')
+    s = schedule('schedule.test')
     tag = uuid()
     f = Task.objects.create(
         id=tag[1],
@@ -25,13 +26,13 @@ def test_admin_views(admin_client):
     t = Task.objects.create(
         id=tag[1],
         name=tag[0],
-        func='test.succes',
+        func='test.success',
         started=timezone.now(),
         stopped=timezone.now(),
         success=True)
     q = OrmQ.objects.create(
         key='test',
-        payload='test')
+        payload=SignedPackage.dumps({'id': 1, 'func': 'test', 'name': 'test'}))
     admin_urls = (
         # schedule
         reverse('admin:django_q_schedule_changelist'),

django_q/tests/test_brokers.py

@@ -18,10 +18,19 @@ def test_broker():
     broker.acknowledge('test')
     broker.ping()
     broker.info()
+    # stats
     assert broker.get_stat('test_1') is None
     broker.set_stat('test_1', 'test', 3)
     assert broker.get_stat('test_1') == 'test'
     assert broker.get_stats('test:*')[0] == 'test'
+    # stats with no cache
+    Conf.CACHE = 'not_configured'
+    broker.cache = broker.get_cache()
+    assert broker.get_stat('test_1') is None
+    broker.set_stat('test_1', 'test', 3)
+    assert broker.get_stat('test_1') is None
+    assert broker.get_stats('test:*') is None
+    Conf.CACHE = 'default'
 
 
 def test_redis():
@@ -49,7 +58,7 @@ def test_disque():
     broker.enqueue('test')
     assert broker.queue_size() == 1
     # dequeue
-    task = broker.dequeue()
+    task = broker.dequeue()[0]
     assert task[1] == 'test'
     broker.acknowledge(task[0])
     assert broker.queue_size() == 0
@@ -61,7 +70,7 @@ def test_disque():
     assert broker.queue_size() == 0
     sleep(1.5)
     assert broker.queue_size() == 1
-    task = broker.dequeue()
+    task = broker.dequeue()[0]
     assert broker.queue_size() == 0
     broker.acknowledge(task[0])
     sleep(1.5)
@@ -78,8 +87,8 @@ def test_disque():
         broker.enqueue('test')
     Conf.BULK = 5
     Conf.DISQUE_FASTACK = True
-    for i in range(5):
+    tasks = broker.dequeue()
-        task = broker.dequeue()
+    for task in tasks:
         assert task is not None
         broker.acknowledge(task[0])
     # test duplicate acknowledge
@@ -117,7 +126,7 @@ def test_ironmq():
     # enqueue
     broker.enqueue('test')
     # dequeue
-    task = broker.dequeue()
+    task = broker.dequeue()[0]
     assert task[1] == 'test'
     broker.acknowledge(task[0])
     assert broker.dequeue() is None
@@ -126,7 +135,7 @@ def test_ironmq():
     broker.enqueue('test')
     assert broker.dequeue() is not None
     sleep(1.5)
-    task = broker.dequeue()
+    task = broker.dequeue()[0]
     assert len(task) > 0
     broker.acknowledge(task[0])
     sleep(1.5)
@@ -141,8 +150,8 @@ def test_ironmq():
     for i in range(5):
         broker.enqueue('test')
     Conf.BULK = 5
-    for i in range(5):
+    tasks = broker.dequeue()
-        task = broker.dequeue()
+    for task in tasks:
         assert task is not None
         broker.acknowledge(task[0])
     # duplicate acknowledge
@@ -174,7 +183,7 @@ def test_sqs():
     # enqueue
     broker.enqueue('test')
     # dequeue
-    task = broker.dequeue()
+    task = broker.dequeue()[0]
     assert task[1] == 'test'
     broker.acknowledge(task[0])
     assert broker.dequeue() is None
@@ -183,26 +192,26 @@ def test_sqs():
     broker.enqueue('test')
     assert broker.dequeue() is not None
     sleep(1.5)
-    task = broker.dequeue()
+    task = broker.dequeue()[0]
     assert len(task) > 0
     broker.acknowledge(task[0])
     sleep(1.5)
     # delete job
     broker.enqueue('test')
-    task_id = broker.dequeue()[0]
+    task_id = broker.dequeue()[0][0]
     broker.delete(task_id)
     assert broker.dequeue() is None
     # fail
     broker.enqueue('test')
     while task is None:
-        task = broker.dequeue()
+        task = broker.dequeue()[0]
     broker.fail(task[0])
     # bulk test
     for i in range(10):
         broker.enqueue('test')
     Conf.BULK = 12
-    for i in range(10):
+    tasks = broker.dequeue()
-        task = broker.dequeue()
+    for task in tasks:
         assert task is not None
         broker.acknowledge(task[0])
     # duplicate acknowledge
@@ -216,6 +225,7 @@ def test_sqs():
     Conf.BULK = 1
     Conf.DJANGO_REDIS = 'default'
 
+
 @pytest.mark.django_db
 def test_orm():
     Conf.ORM = 'default'
@@ -229,7 +239,7 @@ def test_orm():
     broker.enqueue('test')
     assert broker.queue_size() == 1
     # dequeue
-    task = broker.dequeue()
+    task = broker.dequeue()[0]
     assert task[1] == 'test'
     broker.acknowledge(task[0])
     assert broker.queue_size() == 0
@@ -241,7 +251,7 @@ def test_orm():
     assert broker.queue_size() == 0
     sleep(1.5)
     assert broker.queue_size() == 1
-    task = broker.dequeue()
+    task = broker.dequeue()[0]
     assert broker.queue_size() == 0
     broker.acknowledge(task[0])
     sleep(1.5)
@@ -257,8 +267,8 @@ def test_orm():
     for i in range(5):
         broker.enqueue('test')
     Conf.BULK = 5
-    for i in range(5):
+    tasks = broker.dequeue()
-        task = broker.dequeue()
+    for task in tasks:
         assert task is not None
         broker.acknowledge(task[0])
     # test duplicate acknowledge

django_q/tests/test_cluster.py

@@ -215,13 +215,19 @@ def test_async(broker, admin_user):
     assert result(result_j.name) == result_j.result
     # groups
     assert result_group('test_j')[0] == result_j.result
+    assert result_j.group_result()[0] == result_j.result
     assert result_group('test_j', failures=True)[0] == result_j.result
+    assert result_j.group_result(failures=True)[0] == result_j.result
     assert fetch_group('test_j')[0].id == [result_j][0].id
     assert fetch_group('test_j', failures=False)[0].id == [result_j][0].id
     assert count_group('test_j') == 1
+    assert result_j.group_count() == 1
     assert count_group('test_j', failures=True) == 0
+    assert result_j.group_count(failures=True) == 0
     assert delete_group('test_j') == 1
+    assert result_j.group_delete() == 0
     assert delete_group('test_j', tasks=True) is None
+    assert result_j.group_delete(tasks=True) is None
     # task k should not have been saved
     assert fetch(k) is None
     broker.delete_queue()

django_q/tests/test_commands.py

@@ -15,3 +15,4 @@ def test_qmonitor():
 @pytest.mark.django_db
 def test_qinfo():
     call_command('qinfo')
+    call_command('qinfo', config=True)

docs/brokers.rst

@@ -96,11 +96,11 @@ You can override this class if you want to contribute and support your own broke
 
    .. py:method:: enqueue(task)
 
-      Sends a task package to the broker queue and returns a tracking id.
+      Sends a task package to the broker queue and returns a tracking id if available.
 
    .. py:method:: dequeue()
 
-      Gets a task package from the broker and returns a tuple with a tracking id and the package.
+      Gets packages from the broker and returns a list of tuples with a tracking id and the package.
 
    .. py:method:: acknowledge(id)
 

docs/examples.rst

@@ -220,8 +220,8 @@ Adapted from `Sebastian Raschka's blog <http://sebastianraschka.com/Articles/201
 
     # wait for 100 results to return and print it.
     def parzen_hook(task):
-        if count_group('parzen') == 100:
+        if task.group_count() == 100:
-            print(result_group('parzen'))
+            print(task.group_result())
 
 
 Django Q is not optimized for distributed computing, but this example will give you an idea of what you can do with task :ref:`groups`.

docs/monitor.rst

@@ -11,6 +11,10 @@ Start the monitor with Django's `manage.py` command::
 
 .. image:: _static/monitor.png
 
+For all broker types except the Redis broker, the monitor utilizes Django's cache framework to store statistics of running clusters.
+This can be any type of cache backend as long as it can be shared among Django instances. For this reason, the local memory backend will not work.
+
+
 Legend
 ------
 
@@ -88,6 +92,13 @@ Average execution time (`Avg time`) is calculated in seconds over the last 24 ho
 
 Since some of these numbers are based on what is available in your tasks database, limiting or disabling the result backend will skew them.
 
+Like with the monitor, these statistics come from a Redis server or Django's cache framework. So make sure you have either one configured.
+
+To print out the current configuration run::
+
+    $ python manage.py qinfo --config
+
+
 Status
 ------
 

docs/tasks.rst

@@ -127,8 +127,20 @@ Getting results by using :func:`result_group` is of course much faster than usin
 
 .. note::
 
-   Although :func:`fetch_group` returns a queryset, due to the nature of the PickleField , calling ``Queryset.values`` on it will return a list of encoded results.
+   Calling ``Queryset.values`` for the result on Django 1.7 or lower will return a list of encoded results.
-   Use list comprehension or an iterator instead.
+   If you can't upgrade to Django 1.8, use list comprehension or an iterator to return decoded results.
+
+You can also access group functions from a task result instance:
+
+.. code-block:: python
+
+    from django_q import fetch
+
+    task = fetch('winter-speaker-alpha-ceiling')
+    if  task.group_count() > 100:
+        print(task.group_result())
+        task.group_delete()
+        print('Deleted group {}'.format(task.group))
 
 Synchronous testing
 -------------------
@@ -318,6 +330,21 @@ Reference
 
             Time taken represents the time a task spends in the cluster, this includes any time it may have waited in the queue.
 
+    .. py:method:: group_result(failures=False)
+
+    Returns a list of results from this task's group.
+    Set failures to ``True`` to include failed results.
+
+    .. py:method:: group_count(failures=False)
+
+    Returns a count of the number of task results in this task's group.
+    Returns the number of failures when ``failures=True``
+
+    .. py:method:: group_delete(tasks=False)
+
+    Resets the group label on all the tasks in this task's group.
+    If ``tasks=True`` it will also delete the tasks in this group from the database, including itself.
+
     .. py:classmethod:: get_result(task_id)
 
     Gets a result directly by task uuid or name.

requirements.txt

@@ -7,7 +7,7 @@
 arrow==0.6.0
 blessed==1.9.5
 boto3==1.1.3
-botocore==1.2.1           # via boto3
+botocore==1.2.2           # via boto3
 django-picklefield==0.3.2
 django-redis==4.2.0
 docutils==0.12            # via botocore