Restructured all the modules

The core module was becoming too large.
By splitting it in  cluster, monitor and tasks it becomes more manageable. It also reflects the documentation structure.

django_q/__init__.py

@@ -1,29 +1,6 @@
-from django_q.models import Task, Schedule
+from .tasks import async, schedule, result, fetch
-from django_q.core import async, schedule
+from .models import Task, Schedule
 
 VERSION = (0, 1, 4)
 
 default_app_config = 'django_q.apps.DjangoQConfig'
-
-
-def result(name):
-    """
-    Returns the result of the named task
-    :type name: str or unicode
-    :param name: the task name
-    :return: the result object of this task
-    :rtype: object or str
-    """
-    return Task.get_result(name)
-
-
-def get_task(name):
-    """
-    Returns the processed task
-    :param name: the task name
-    :type name: str or unicode
-    :return: the full task object
-    :rtype: Task
-    """
-    if Task.objects.filter(name=name).exists():
-        return Task.objects.get(name=name)

django_q/admin.py

@@ -1,6 +1,6 @@
 from django.contrib import admin
-
+from django.utils.translation import ugettext_lazy as _
-from django_q.core import async
+from .tasks import async
 from .models import Success, Failure, Schedule
 
 
@@ -35,12 +35,12 @@ def retry_failed(FailAdmin, request, queryset):
         task.delete()
 
 
-retry_failed.short_description = "Resubmit selected tasks to Q"
+retry_failed.short_description = _("Resubmit selected tasks to queue")
 
 
 class FailAdmin(admin.ModelAdmin):
     list_display = (
-        u'name',
+        'name',
         'func',
         'started',
         'result'
@@ -61,7 +61,7 @@ class FailAdmin(admin.ModelAdmin):
 
 class ScheduleAdmin(admin.ModelAdmin):
     list_display = (
-        u'id',
+        'id',
         'func',
         'schedule_type',
         'repeats',

django_q/cluster.py

@@ -3,53 +3,39 @@ from __future__ import unicode_literals
 from __future__ import print_function
 from __future__ import division
 from __future__ import absolute_import
-import ast
 from builtins import range
-
 from future import standard_library
 
 standard_library.install_aliases()
 
 # Standard
 import importlib
-import logging
 import os
 import signal
-from multiprocessing import Queue, Event, Process, Value, current_process
 import socket
 import sys
+import ast
 from time import sleep
+from multiprocessing import Queue, Event, Process, Value, current_process
 
 try:
     import cPickle as pickle
 except ImportError:
     import pickle
 
-# External
+# external
-import redis
 import arrow
 
 # Django
 from django.core import signing
 from django.utils import timezone
+from django.utils.translation import ugettext_lazy as _
 
 # Local
-from .conf import Conf
+from .conf import Conf, redis_client, logger
-from .humanhash import uuid
 from .models import Task, Success, Schedule
-
+from .monitor import Status, Stat
-logger = logging.getLogger('django-q')
+from .tasks import SignedPackage, async
-
-# Set up standard logging handler in case there is none
-if not logger.handlers:
-    logger.setLevel(level=getattr(logging, Conf.LOG_LEVEL))
-    formatter = logging.Formatter(fmt='%(asctime)s [Q] %(levelname)s %(message)s',
-                                  datefmt='%H:%M:%S')
-    handler = logging.StreamHandler()
-    handler.setFormatter(formatter)
-    logger.addHandler(handler)
-
-redis_client = redis.StrictRedis(**Conf.REDIS)
 
 
 class Cluster(object):
@@ -82,7 +68,7 @@ class Cluster(object):
         self.start_event = Event()
         self.sentinel = Process(target=Sentinel, args=(self.stop_event, self.start_event, self.list_key, self.timeout))
         self.sentinel.start()
-        logger.info('Q Cluster-{} starting.'.format(self.pid))
+        logger.info(_('Q Cluster-{} starting.').format(self.pid))
         while not self.start_event.is_set():
             sleep(0.2)
         return self.pid
@@ -90,16 +76,16 @@ class Cluster(object):
     def stop(self):
         if not self.sentinel.is_alive():
             return False
-        logger.info('Q Cluster-{} stopping.'.format(self.pid))
+        logger.info(_('Q Cluster-{} stopping.').format(self.pid))
         self.stop_event.set()
         self.sentinel.join()
-        logger.info('Q Cluster-{} has stopped.'.format(self.pid))
+        logger.info(_('Q Cluster-{} has stopped.').format(self.pid))
         self.start_event = None
         self.stop_event = None
         return True
 
     def sig_handler(self, signum, frame):
-        logger.debug('{} got signal {}'.format(current_process().name, Conf.SIGNAL_NAMES.get(signum, 'UNKNOWN')))
+        logger.debug(_('{} got signal {}').format(current_process().name, Conf.SIGNAL_NAMES.get(signum, 'UNKNOWN')))
         self.stop()
 
     @property
@@ -201,17 +187,17 @@ class Sentinel(object):
         process.terminate()
         if process == self.monitor:
             self.monitor = self.spawn_monitor()
-            logger.error("reincarnated monitor {} after sudden.".format(process.pid))
+            logger.error(_("reincarnated monitor {} after sudden.").format(process.pid))
         elif process == self.pusher:
             self.pusher = self.spawn_pusher()
-            logger.error("reincarnated pusher {} after sudden death".format(process.pid))
+            logger.error(_("reincarnated pusher {} after sudden death.").format(process.pid))
         else:
             self.pool.remove(process)
             self.spawn_worker()
             if int(process.timer.value) >= self.timeout:
-                logger.warn("reincarnated worker {} after timeout.".format(process.pid))
+                logger.warn(_("reincarnated worker {} after timeout.").format(process.pid))
             else:
-                logger.error("reincarnated worker {} after sudden death.".format(process.pid))
+                logger.error(_("reincarnated worker {} after sudden death.").format(process.pid))
         self.reincarnations += 1
 
     def spawn_cluster(self):
@@ -222,10 +208,10 @@ class Sentinel(object):
         self.pusher = self.spawn_pusher()
 
     def guard(self):
-        logger.info('{} guarding cluster at {}'.format(current_process().name, self.pid))
+        logger.info(_('{} guarding cluster at {}').format(current_process().name, self.pid))
         self.start_event.set()
         Stat(self).save()
-        logger.info('Q Cluster-{} running.'.format(self.parent_pid))
+        logger.info(_('Q Cluster-{} running.').format(self.parent_pid))
         scheduler(list_key=self.list_key)
         counter = 0
         # Guard loop. Runs at least once
@@ -258,13 +244,14 @@ class Sentinel(object):
     def stop(self):
         Stat(self).save()
         name = current_process().name
-        logger.info('{} stopping pool processes'.format(name))
+        logger.info('{} stopping cluster processes'.format(name))
         # Stopping pusher
         self.event_out.set()
+        # Wait for it to stop
         while self.pusher.is_alive():
             sleep(0.2)
             Stat(self).save()
-        # Putting poison pills in the queue
+        # Put poison pills in the queue
         for _ in range(self.pool_size):
             self.task_queue.put('STOP')
         # Wait for all the workers to exit
@@ -290,16 +277,16 @@ def pusher(task_queue, e, list_key=Conf.Q_LIST, r=redis_client):
     :type e: multiprocessing.Event
     :type list_key: str
     """
-    logger.info('{} pushing tasks at {}'.format(current_process().name, current_process().pid))
+    logger.info(_('{} pushing tasks at {}').format(current_process().name, current_process().pid))
     while True:
         task = r.blpop(list_key, 1)
         if task:
             task = task[1]
             task_queue.put(task)
-            logger.debug('queueing {}'.format(task))
+            logger.debug(_('queueing {}').format(task))
         if e.is_set():
             break
-    logger.info("{} stopped pushing tasks".format(current_process().name))
+    logger.info(_("{} stopped pushing tasks.").format(current_process().name))
 
 
 def monitor(done_queue):
@@ -308,14 +295,14 @@ def monitor(done_queue):
     :type done_queue: multiprocessing.Queue
     """
     name = current_process().name
-    logger.info("{} monitoring at {}".format(name, current_process().pid))
+    logger.info(_("{} monitoring at {}").format(name, current_process().pid))
     for task in iter(done_queue.get, 'STOP'):
         if task['success']:
-            logger.info("Processed [{}]".format(task['name']))
+            logger.info(_("Processed [{}]").format(task['name']))
         else:
-            logger.error("Failed [{}] - {}".format(task['name'], task['result']))
+            logger.error(_("Failed [{}] - {}").format(task['name'], task['result']))
         save_task(task)
-    logger.info("{} stopped monitoring results".format(name))
+    logger.info(_("{} stopped monitoring results").format(name))
 
 
 def worker(task_queue, done_queue, timer):
@@ -326,7 +313,7 @@ def worker(task_queue, done_queue, timer):
     :type timer: multiprocessing.Value
     """
     name = current_process().name
-    logger.info('{} ready for work at {}'.format(name, current_process().pid))
+    logger.info(_('{} ready for work at {}').format(name, current_process().pid))
     task_count = 0
     # Start reading the task queue
     for pack in iter(task_queue.get, 'STOP'):
@@ -340,7 +327,7 @@ def worker(task_queue, done_queue, timer):
             logger.error(e)
             continue
         # Get the function from the task
-        logger.info('{} processing [{}]'.format(name, task['name']))
+        logger.info(_('{} processing [{}]').format(name, task['name']))
         f = task['func']
         # if it's not an instance try to get it from the string
         if not callable(task['func']):
@@ -368,7 +355,7 @@ def worker(task_queue, done_queue, timer):
         # Recycle
         if task_count == Conf.RECYCLE and task_queue.qsize() == 0:
             break
-    logger.info('{} stopped doing work'.format(name))
+    logger.info(_('{} stopped doing work').format(name))
 
 
 def save_task(task):
@@ -396,192 +383,6 @@ def save_task(task):
         logger.exception(e)
 
 
-def async(func, *args, **kwargs):
-    """
-    Sends a task to the cluster
-    """
-
-    hook = kwargs.pop('hook', None)
-    list_key = kwargs.pop('list_key', Conf.Q_LIST)
-    r = kwargs.pop('redis', redis_client)
-
-    task = {'name': uuid()[0], 'func': func, 'hook': hook, 'args': args, 'kwargs': kwargs, 'started': timezone.now()}
-    pack = SignedPackage.dumps(task)
-    r.rpush(list_key, pack)
-    logger.debug('Pushed {}'.format(pack))
-    return task['name']
-
-
-class SignedPackage(object):
-    """
-    Wraps Django's signing module with custom Pickle serializer
-    """
-
-    @staticmethod
-    def dumps(obj, compressed=Conf.COMPRESSED):
-        return signing.dumps(obj,
-                             key=Conf.SECRET_KEY,
-                             salt='django_q.q',
-                             compress=compressed,
-                             serializer=PickleSerializer)
-
-    @staticmethod
-    def loads(obj):
-        return signing.loads(obj,
-                             key=Conf.SECRET_KEY,
-                             salt='django_q.q',
-                             serializer=PickleSerializer)
-
-
-class PickleSerializer(object):
-    """
-    Simple wrapper around Pickle for signing.dumps and
-    signing.loads.
-    """
-
-    @staticmethod
-    def dumps(obj):
-        return pickle.dumps(obj)
-
-    @staticmethod
-    def loads(data):
-        return pickle.loads(data)
-
-
-class Status(object):
-    """
-    Cluster status base object
-    """
-
-    def __init__(self, pid):
-        self.workers = []
-        self.tob = None
-        self.reincarnations = 0
-        self.cluster_id = pid
-        self.sentinel = 0
-        self.status = 'Idle'
-        self.done_q_size = 0
-        self.host = socket.gethostname()
-        self.monitor = 0
-        self.task_q_size = 0
-        self.pusher = 0
-        self.timestamp = timezone.now()
-
-
-class Stat(Status):
-    """
-    Status object for Cluster monitoring
-    """
-
-    def __init__(self, sentinel):
-        super(Stat, self).__init__(sentinel.parent_pid)
-        self.r = sentinel.r
-        self.tob = sentinel.tob
-        self.reincarnations = sentinel.reincarnations
-        self.sentinel = sentinel.pid
-        self.status = sentinel.status()
-        self.done_q_size = sentinel.done_queue.qsize()
-        if sentinel.monitor:
-            self.monitor = sentinel.monitor.pid
-        self.task_q_size = sentinel.task_queue.qsize()
-        if sentinel.pusher:
-            self.pusher = sentinel.pusher.pid
-        for w in sentinel.pool:
-            self.workers.append(w.pid)
-
-    def uptime(self):
-        return (timezone.now() - self.tob).total_seconds()
-
-    @property
-    def key(self):
-        """
-        :return: redis key for this cluster statistic
-        """
-        return self.get_key(self.cluster_id)
-
-    @staticmethod
-    def get_key(cluster_id):
-        """
-        :param cluster_id: cluster ID
-        :return: redis key for the cluster statistic
-        """
-        return '{}:{}'.format(Conf.Q_STAT, cluster_id)
-
-    def save(self):
-        self.r.set(self.key, SignedPackage.dumps(self, True), 3)
-
-    def empty_queues(self):
-        return self.done_q_size + self.task_q_size == 0
-
-    @staticmethod
-    def get(cluster_id, r=redis_client):
-        """
-        gets the current status for the cluster
-        :param cluster_id: id of the cluster
-        :return: Stat or Status
-        """
-        key = Stat.get_key(cluster_id)
-        if r.exists(key):
-            pack = r.get(key)
-            try:
-                return SignedPackage.loads(pack)
-            except signing.BadSignature:
-                return None
-        return Status(cluster_id)
-
-    @staticmethod
-    def get_all(r=redis_client):
-        """
-        Gets status for all currently running clusters with the same prefix and secret key
-        :return: Stat list
-        """
-        stats = []
-        keys = r.keys(pattern='{}:*'.format(Conf.Q_STAT))
-        if keys:
-            packs = r.mget(keys)
-            for pack in packs:
-                try:
-                    stats.append(SignedPackage.loads(pack))
-                except signing.BadSignature:
-                    continue
-        return stats
-
-    def __getstate__(self):
-        # Don't pickle the redis connection
-        state = dict(self.__dict__)
-        del state['r']
-        return state
-
-
-def schedule(func, *args, **kwargs):
-    """
-    :param func: function to schedule
-    :param args: function arguments
-    :param hook: optional result hook function
-    :type schedule_type: Schedule.TYPE
-    :param repeats: how many times to repeat. 0=never, -1=always
-    :param next_run: Next scheduled run
-    :type next_run: datetime.datetime
-    :param kwargs: function keyword arguments
-    :return: the schedule object
-    :rtype: Schedule
-    """
-
-    hook = kwargs.pop('hook', None)
-    schedule_type = kwargs.pop('schedule_type', Schedule.ONCE)
-    repeats = kwargs.pop('repeats', -1)
-    next_run = kwargs.pop('next_run', timezone.now())
-
-    return Schedule.objects.create(func=func,
-                                   hook=hook,
-                                   args=args,
-                                   kwargs=kwargs,
-                                   schedule_type=schedule_type,
-                                   repeats=repeats,
-                                   next_run=next_run
-                                   )
-
-
 def scheduler(list_key=Conf.Q_LIST):
     """
     Creates a task from a schedule at the scheduled time and schedules next run
@@ -626,7 +427,7 @@ def scheduler(list_key=Conf.Q_LIST):
         kwargs['list_key'] = list_key
         s.task = async(s.func, *args, **kwargs)
         if not s.task:
-            logger.error('{} failed to create task from  schedule {}').format(current_process().name, s.id)
+            logger.error(_('{} failed to create task from  schedule {}').format(current_process().name, s.id))
         else:
-            logger.info('{} created [{}] from schedule {}'.format(current_process().name, s.task, s.id))
+            logger.info(_('{} created [{}] from schedule {}').format(current_process().name, s.task, s.id))
-        s.save()
+        s.save()

django_q/conf.py

@@ -1,7 +1,10 @@
+import logging
 from signal import signal
 from multiprocessing import cpu_count
 
+from django.utils.translation import ugettext_lazy as _
 from django.conf import settings
+import redis
 
 
 class Conf(object):
@@ -52,9 +55,24 @@ class Conf(object):
     # Getting the signal names
     SIGNAL_NAMES = dict((getattr(signal, n), n) for n in dir(signal) if n.startswith('SIG') and '_' not in n)
 
-    # Cluster status descriptions
+    # Translators: Cluster status descriptions
-    STARTING = 'Starting'
+    STARTING = _('Starting')
-    WORKING = 'Working'
+    WORKING = _('Working')
-    IDLE = "Idle"
+    IDLE = _("Idle")
-    STOPPED = 'Stopped'
+    STOPPED = _('Stopped')
-    STOPPING = 'Stopping'
+    STOPPING = _('Stopping')
+
+# logger
+logger = logging.getLogger('django-q')
+
+# Set up standard logging handler in case there is none
+if not logger.handlers:
+    logger.setLevel(level=getattr(logging, Conf.LOG_LEVEL))
+    formatter = logging.Formatter(fmt='%(asctime)s [Q] %(levelname)s %(message)s',
+                                  datefmt='%H:%M:%S')
+    handler = logging.StreamHandler()
+    handler.setFormatter(formatter)
+    logger.addHandler(handler)
+
+# redis client
+redis_client = redis.StrictRedis(**Conf.REDIS)

django_q/management/commands/qcluster.py

@@ -1,9 +1,11 @@
 from django.core.management.base import BaseCommand
-from django_q.core import Cluster
+from django.utils.translation import ugettext as _
+from django_q.cluster import Cluster
 
 
 class Command(BaseCommand):
-    help = "Starts a Django Q Cluster."
+    # Translators: help text for qcluster management command
+    help = _("Starts a Django Q Cluster.")
 
     def handle(self, *args, **options):
         q = Cluster()

django_q/management/commands/qmonitor.py

@@ -1,84 +1,16 @@
 # Django
 from django.core.management.base import BaseCommand
-from django.utils import timezone
+from django.utils.translation import ugettext as _
 
-# External
-from blessed import Terminal
 
-# Local
+from ...monitor import monitor
-from django_q.core import Stat, redis_client
-from django_q.conf import Conf
 
-# TODO add name argument to monitor different clusters
 
 class Command(BaseCommand):
-    help = "Monitors cluster activity"
+    # Translators: help text for qmonitor management command
+    help = _("Monitors Q Cluster activity")
 
     def handle(self, *args, **options):
         monitor()
 
 
-def monitor(run_once=False):
-    term = Terminal()
-    r = redis_client
-    with term.fullscreen(), term.hidden_cursor(), term.cbreak():
-        val = None
-        start_width = int(term.width / 8)
-        while val not in (u'q', u'Q',):
-            col_width = int(term.width / 8)
-            # In case of resize
-            if col_width != start_width:
-                print(term.clear)
-                start_width = col_width
-            print(term.move(0, 0) + term.black_on_green(term.center('Host', width=col_width - 1)))
-            print(term.move(0, 1 * col_width) + term.black_on_green(term.center('Id', width=col_width - 1)))
-            print(term.move(0, 2 * col_width) + term.black_on_green(term.center('Status', width=col_width - 1)))
-            print(term.move(0, 3 * col_width) + term.black_on_green(term.center('Pool', width=col_width - 1)))
-            print(term.move(0, 4 * col_width) + term.black_on_green(term.center('TQ', width=col_width - 1)))
-            print(term.move(0, 5 * col_width) + term.black_on_green(term.center('RQ', width=col_width - 1)))
-            print(term.move(0, 6 * col_width) + term.black_on_green(term.center('RC', width=col_width - 1)))
-            print(term.move(0, 7 * col_width) + term.black_on_green(term.center('Up', width=col_width - 1)))
-            i = 2
-            stats = Stat.get_all(r=r)
-            print(term.clear_eos())
-            for stat in stats:
-                # color status
-                if stat.status == Conf.WORKING:
-                    status = term.green(Conf.WORKING)
-                elif stat.status == Conf.STOPPED:
-                    status = term.red(Conf.STOPPED)
-                elif stat.status == Conf.IDLE:
-                    status = Conf.IDLE
-                else:
-                    status = term.yellow(stat.status)
-                # color q's
-                tasks = stat.task_q_size
-                if tasks > 0:
-                    tasks = term.cyan(str(tasks))
-                results = stat.done_q_size
-                if results > 0:
-                    results = term.cyan(str(results))
-                # color workers
-                workers = len(stat.workers)
-                if workers < Conf.WORKERS:
-                    workers = term.yellow(str(workers))
-                # format uptime
-                uptime = (timezone.now() - stat.tob).total_seconds()
-                hours, remainder = divmod(uptime, 3600)
-                minutes, seconds = divmod(remainder, 60)
-                uptime = '%d:%02d:%02d' % (hours, minutes, seconds)
-                # print to the terminal
-                print(term.move(i, 0) + term.center(stat.host[:col_width - 1], width=col_width - 1))
-                print(term.move(i, 1 * col_width) + term.center(stat.cluster_id, width=col_width - 1))
-                print(term.move(i, 2 * col_width) + term.center(status, width=col_width - 1))
-                print(term.move(i, 3 * col_width) + term.center(workers, width=col_width - 1))
-                print(term.move(i, 4 * col_width) + term.center(tasks, width=col_width - 1))
-                print(term.move(i, 5 * col_width) + term.center(results, width=col_width - 1))
-                print(term.move(i, 6 * col_width) + term.center(stat.reincarnations, width=col_width - 1))
-                print(term.move(i, 7 * col_width) + term.center(uptime, width=col_width - 1))
-                i += 1
-            # for testing
-            if run_once:
-                return Stat.get_all(r=r)
-            print(term.move(i + 2, 0) + term.center('[Press q to quit]'))
-            val = term.inkey(timeout=1)

django_q/models.py

@@ -67,6 +67,7 @@ class Success(Task):
     class Meta:
         app_label = 'django_q'
         verbose_name = _('Successful task')
+        verbose_name_plural = _('Successful tasks')
         proxy = True
 
 
@@ -82,6 +83,7 @@ class Failure(Task):
     class Meta:
         app_label = 'django_q'
         verbose_name = _('Failed task')
+        verbose_name_plural = _('Failed tasks')
         proxy = True
 
 
@@ -135,4 +137,5 @@ class Schedule(models.Model):
     class Meta:
         app_label = 'django_q'
         verbose_name = _('Scheduled task')
+        verbose_name_plural = _('Scheduled tasks')
         ordering = ['next_run']

django_q/monitor.py

@@ -0,0 +1,184 @@
+import socket
+
+# external
+from blessed import Terminal
+
+# django
+from django.core import signing
+from django.utils import timezone
+from django.utils.translation import ugettext as _
+
+# local
+from .conf import Conf, redis_client
+from .tasks import SignedPackage
+
+
+def monitor(run_once=False):
+    term = Terminal()
+    r = redis_client
+    with term.fullscreen(), term.hidden_cursor(), term.cbreak():
+        val = None
+        start_width = int(term.width / 8)
+        while val not in (u'q', u'Q',):
+            col_width = int(term.width / 8)
+            # In case of resize
+            if col_width != start_width:
+                print(term.clear)
+                start_width = col_width
+            print(term.move(0, 0) + term.black_on_green(term.center(_('Host'), width=col_width - 1)))
+            print(term.move(0, 1 * col_width) + term.black_on_green(term.center(_('Id'), width=col_width - 1)))
+            print(term.move(0, 2 * col_width) + term.black_on_green(term.center(_('State'), width=col_width - 1)))
+            print(term.move(0, 3 * col_width) + term.black_on_green(term.center(_('Pool'), width=col_width - 1)))
+            print(term.move(0, 4 * col_width) + term.black_on_green(term.center(_('TQ'), width=col_width - 1)))
+            print(term.move(0, 5 * col_width) + term.black_on_green(term.center(_('RQ'), width=col_width - 1)))
+            print(term.move(0, 6 * col_width) + term.black_on_green(term.center(_('RC'), width=col_width - 1)))
+            print(term.move(0, 7 * col_width) + term.black_on_green(term.center(_('Up'), width=col_width - 1)))
+            i = 2
+            stats = Stat.get_all(r=r)
+            print(term.clear_eos())
+            for stat in stats:
+                # color status
+                if stat.status == Conf.WORKING:
+                    status = term.green(Conf.WORKING)
+                elif stat.status == Conf.STOPPED:
+                    status = term.red(Conf.STOPPED)
+                elif stat.status == Conf.IDLE:
+                    status = Conf.IDLE
+                else:
+                    status = term.yellow(stat.status)
+                # color q's
+                tasks = stat.task_q_size
+                if tasks > 0:
+                    tasks = term.cyan(str(tasks))
+                results = stat.done_q_size
+                if results > 0:
+                    results = term.cyan(str(results))
+                # color workers
+                workers = len(stat.workers)
+                if workers < Conf.WORKERS:
+                    workers = term.yellow(str(workers))
+                # format uptime
+                uptime = (timezone.now() - stat.tob).total_seconds()
+                hours, remainder = divmod(uptime, 3600)
+                minutes, seconds = divmod(remainder, 60)
+                uptime = '%d:%02d:%02d' % (hours, minutes, seconds)
+                # print to the terminal
+                print(term.move(i, 0) + term.center(stat.host[:col_width - 1], width=col_width - 1))
+                print(term.move(i, 1 * col_width) + term.center(stat.cluster_id, width=col_width - 1))
+                print(term.move(i, 2 * col_width) + term.center(status, width=col_width - 1))
+                print(term.move(i, 3 * col_width) + term.center(workers, width=col_width - 1))
+                print(term.move(i, 4 * col_width) + term.center(tasks, width=col_width - 1))
+                print(term.move(i, 5 * col_width) + term.center(results, width=col_width - 1))
+                print(term.move(i, 6 * col_width) + term.center(stat.reincarnations, width=col_width - 1))
+                print(term.move(i, 7 * col_width) + term.center(uptime, width=col_width - 1))
+                i += 1
+            # for testing
+            if run_once:
+                return Stat.get_all(r=r)
+            print(term.move(i + 2, 0) + term.center(_('[Press q to quit]')))
+            val = term.inkey(timeout=1)
+
+
+class Status(object):
+    """
+    Cluster status base class
+    """
+
+    def __init__(self, pid):
+        self.workers = []
+        self.tob = None
+        self.reincarnations = 0
+        self.cluster_id = pid
+        self.sentinel = 0
+        self.status = 'Idle'
+        self.done_q_size = 0
+        self.host = socket.gethostname()
+        self.monitor = 0
+        self.task_q_size = 0
+        self.pusher = 0
+        self.timestamp = timezone.now()
+
+
+class Stat(Status):
+    """
+    Status object for Cluster monitoring
+    """
+
+    def __init__(self, sentinel):
+        super(Stat, self).__init__(sentinel.parent_pid)
+        self.r = sentinel.r
+        self.tob = sentinel.tob
+        self.reincarnations = sentinel.reincarnations
+        self.sentinel = sentinel.pid
+        self.status = sentinel.status()
+        self.done_q_size = sentinel.done_queue.qsize()
+        if sentinel.monitor:
+            self.monitor = sentinel.monitor.pid
+        self.task_q_size = sentinel.task_queue.qsize()
+        if sentinel.pusher:
+            self.pusher = sentinel.pusher.pid
+        for w in sentinel.pool:
+            self.workers.append(w.pid)
+
+    def uptime(self):
+        return (timezone.now() - self.tob).total_seconds()
+
+    @property
+    def key(self):
+        """
+        :return: redis key for this cluster statistic
+        """
+        return self.get_key(self.cluster_id)
+
+    @staticmethod
+    def get_key(cluster_id):
+        """
+        :param cluster_id: cluster ID
+        :return: redis key for the cluster statistic
+        """
+        return '{}:{}'.format(Conf.Q_STAT, cluster_id)
+
+    def save(self):
+        self.r.set(self.key, SignedPackage.dumps(self, True), 3)
+
+    def empty_queues(self):
+        return self.done_q_size + self.task_q_size == 0
+
+    @staticmethod
+    def get(cluster_id, r=redis_client):
+        """
+        gets the current status for the cluster
+        :param cluster_id: id of the cluster
+        :return: Stat or Status
+        """
+        key = Stat.get_key(cluster_id)
+        if r.exists(key):
+            pack = r.get(key)
+            try:
+                return SignedPackage.loads(pack)
+            except signing.BadSignature:
+                return None
+        return Status(cluster_id)
+
+    @staticmethod
+    def get_all(r=redis_client):
+        """
+        Gets status for all currently running clusters with the same prefix and secret key
+        :return: list of type Stat
+        """
+        stats = []
+        keys = r.keys(pattern='{}:*'.format(Conf.Q_STAT))
+        if keys:
+            packs = r.mget(keys)
+            for pack in packs:
+                try:
+                    stats.append(SignedPackage.loads(pack))
+                except signing.BadSignature:
+                    continue
+        return stats
+
+    def __getstate__(self):
+        # Don't pickle the redis connection
+        state = dict(self.__dict__)
+        del state['r']
+        return state

django_q/tasks.py

@@ -0,0 +1,119 @@
+try:
+    import cPickle as pickle
+except ImportError:
+    import pickle
+
+# django
+from django.core import signing
+from django.utils import timezone
+
+# local
+from .conf import Conf, redis_client, logger
+from .models import Schedule, Task
+from .humanhash import uuid
+
+
+
+
+def async(func, *args, **kwargs):
+    """
+    Sends a task to the cluster
+    """
+
+    hook = kwargs.pop('hook', None)
+    list_key = kwargs.pop('list_key', Conf.Q_LIST)
+    r = kwargs.pop('redis', redis_client)
+
+    task = {'name': uuid()[0], 'func': func, 'hook': hook, 'args': args, 'kwargs': kwargs, 'started': timezone.now()}
+    pack = SignedPackage.dumps(task)
+    r.rpush(list_key, pack)
+    logger.debug('Pushed {}'.format(pack))
+    return task['name']
+
+
+def schedule(func, *args, **kwargs):
+    """
+    :param func: function to schedule
+    :param args: function arguments
+    :param hook: optional result hook function
+    :type schedule_type: Schedule.TYPE
+    :param repeats: how many times to repeat. 0=never, -1=always
+    :param next_run: Next scheduled run
+    :type next_run: datetime.datetime
+    :param kwargs: function keyword arguments
+    :return: the schedule object
+    :rtype: Schedule
+    """
+
+    hook = kwargs.pop('hook', None)
+    schedule_type = kwargs.pop('schedule_type', Schedule.ONCE)
+    repeats = kwargs.pop('repeats', -1)
+    next_run = kwargs.pop('next_run', timezone.now())
+
+    return Schedule.objects.create(func=func,
+                                   hook=hook,
+                                   args=args,
+                                   kwargs=kwargs,
+                                   schedule_type=schedule_type,
+                                   repeats=repeats,
+                                   next_run=next_run
+                                   )
+
+
+def result(name):
+    """
+    Returns the result of the named task
+    :type name: str or unicode
+    :param name: the task name
+    :return: the result object of this task
+    :rtype: object or str
+    """
+    return Task.get_result(name)
+
+
+def fetch(name):
+    """
+    Returns the processed task
+    :param name: the task name
+    :type name: str or unicode
+    :return: the full task object
+    :rtype: Task
+    """
+    if Task.objects.filter(name=name).exists():
+        return Task.objects.get(name=name)
+
+
+class SignedPackage(object):
+    """
+    Wraps Django's signing module with custom Pickle serializer
+    """
+
+    @staticmethod
+    def dumps(obj, compressed=Conf.COMPRESSED):
+        return signing.dumps(obj,
+                             key=Conf.SECRET_KEY,
+                             salt='django_q.q',
+                             compress=compressed,
+                             serializer=PickleSerializer)
+
+    @staticmethod
+    def loads(obj):
+        return signing.loads(obj,
+                             key=Conf.SECRET_KEY,
+                             salt='django_q.q',
+                             serializer=PickleSerializer)
+
+
+class PickleSerializer(object):
+    """
+    Simple wrapper around Pickle for signing.dumps and
+    signing.loads.
+    """
+
+    @staticmethod
+    def dumps(obj):
+        return pickle.dumps(obj)
+
+    @staticmethod
+    def loads(data):
+        return pickle.loads(data)

django_q/tests/test_admin.py

@@ -4,8 +4,8 @@ from django.utils import timezone
 
 import pytest
 
-from django_q import Task, schedule
+from django_q.tasks import schedule
-from django_q.models import Failure
+from django_q.models import Task, Failure
 
 
 @pytest.mark.django_db

django_q/tests/test_cluster.py

@@ -8,11 +8,12 @@ import pytest
 myPath = os.path.dirname(os.path.abspath(__file__))
 sys.path.insert(0, myPath + '/../')
 
-from django_q.core import Cluster, async, pusher, worker, monitor, redis_client, Sentinel
+from django_q.cluster import Cluster, Sentinel, pusher, worker, monitor
 from django_q.humanhash import DEFAULT_WORDLIST
-from django_q import result, get_task, Task
+from django_q.tasks import fetch, async, result
-from django_q.tests.tasks import multiply
+from django_q.models import Task
-from django_q.conf import Conf
+from django_q.conf import Conf, redis_client
+from .tasks import multiply
 
 
 class WordClass(object):
@@ -143,40 +144,40 @@ def test_async(r):
     assert result_queue.qsize() == 0
     # Check the results
     # task a
-    result_a = get_task(a)
+    result_a = fetch(a)
     assert result_a is not None
     assert result_a.success is True
     assert result(a) == 1506
     # task b
-    result_b = get_task(b)
+    result_b = fetch(b)
     assert result_b is not None
     assert result_b.success is True
     assert result(b) == 1506
     # task c
-    result_c = get_task(c)
+    result_c = fetch(c)
     assert result_c is not None
     assert result_c.success is False
     # task d
-    result_d = get_task(d)
+    result_d = fetch(d)
     assert result_d is not None
     assert result_d.success is False
     # task e
-    result_e = get_task(e)
+    result_e = fetch(e)
     assert result_e is not None
     assert result_e.success is True
     assert result(e) is None
     # task f
-    result_f = get_task(f)
+    result_f = fetch(f)
     assert result_f is not None
     assert result_f.success is True
     assert result(f) == 1506
     # task g
-    result_g = get_task(g)
+    result_g = fetch(g)
     assert result_g is not None
     assert result_g.success is True
     assert result(g) == 'John'
     # task h
-    result_h = get_task(h)
+    result_h = fetch(h)
     assert result_h is not None
     assert result_h.success is True
     assert result(h) == 12

django_q/tests/test_monitor.py

@@ -1,5 +1,5 @@
-from django_q.core import Cluster
+from django_q.cluster import Cluster
-from django_q.management.commands.qmonitor import monitor
+from django_q.monitor import monitor
 
 
 def test_monitor():

django_q/tests/test_scheduler.py

@@ -2,8 +2,9 @@ from multiprocessing import Queue, Event, Value
 
 import pytest
 
-from django_q.core import scheduler, pusher, worker, monitor, redis_client, schedule as create_schedule
+from django_q.conf import redis_client
-from django_q import Schedule, get_task
+from django_q.cluster import pusher, worker, monitor, scheduler
+from django_q.tasks import Schedule, fetch, schedule as create_schedule
 
 
 @pytest.fixture
@@ -44,7 +45,7 @@ def test_scheduler(r):
     assert schedule.repeats == 0
     assert schedule.last_run() is not None
     assert schedule.success() is True
-    task = get_task(schedule.task)
+    task = fetch(schedule.task)
     assert task is not None
     assert task.success is True
     assert task.result < 0