cached result backend

first version of a result backend using django's cache framework

django_q/cluster.py

@@ -325,7 +325,11 @@ def monitor(result_queue, broker=None):
         if ack_id:
             broker.acknowledge(ack_id)
         # save the result
-        save_task(task)
+        if task.get('cached', False):
+            save_cache(task, broker)
+        else:
+            save_task(task)
+        # log the result
         if task['success']:
             logger.info(_("Processed [{}]").format(task['name']))
         else:
@@ -384,7 +388,7 @@ def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT):
 
 def save_task(task):
     """
-    Saves the task package to Django
+    Saves the task package to Django or the cache
     """
     # SAVE LIMIT < 0 : Don't save success
     if not task.get('save', Conf.SAVE_LIMIT > 0) and task['success']:
@@ -409,6 +413,26 @@ def save_task(task):
         logger.error(e)
 
 
+def save_cache(task, broker):
+    key = 'django_q:{}:results'.format(broker.list_key)
+    timeout = task['cached']
+    if timeout is True:
+        timeout = None
+    try:
+        task_package = signing.SignedPackage.dumps(task)
+        group = task.get('group', False)
+        if group:
+            group_list = broker.cache.get('{}:{}'.format(key, group)) or []
+            group_list.append(task_package)
+            broker.cache.set('{}:{}'.format(key, group), group_list, timeout)
+        else:
+            broker.cache.set('{}:{}'.format(key, task['id']),
+                             task_package,
+                             timeout)
+    except Exception as e:
+        logger.error(e)
+
+
 def scheduler(broker=None):
     """
     Creates a task from a schedule at the scheduled time and schedules next run

django_q/conf.py

@@ -113,6 +113,10 @@ class Conf(object):
     # The Django cache to use
     CACHE = conf.get('cache', 'default')
 
+    # Use the cache as result backend. Can be 'True' or an integer representing the global cache timeout.
+    # i.e 'cached: 60' , will make all results go the cache and expire in 60 seconds.
+    CACHED = conf.get('cached', False)
+
     # If set to False the scheduler won't execute tasks in the past.
     # Instead it will run once and reschedule the next run in the future. Defaults to True.
     CATCH_UP = conf.get('catch_up', True)

django_q/tasks.py

@@ -23,6 +23,7 @@ def async(func, *args, **kwargs):
     sync = options.pop('sync', False)
     group = options.pop('group', None)
     save = options.pop('save', None)
+    cached = options.pop('cached', Conf.CACHED)
     # get an id
     tag = uuid()
     # build the task package
@@ -38,6 +39,8 @@ def async(func, *args, **kwargs):
         task['group'] = group
     if save is not None:
         task['save'] = save
+    if cached:
+        task['cached'] = cached
     # sign it
     pack = signing.SignedPackage.dumps(task)
     if sync or Conf.SYNC:
@@ -83,7 +86,7 @@ def schedule(func, *args, **kwargs):
                                    )
 
 
-def result(task_id, wait=0):
+def result(task_id, wait=0, cached=Conf.CACHED):
     """
     Return the result of the named task.
 
@@ -94,6 +97,8 @@ def result(task_id, wait=0):
     :return: the result object of this task
     :rtype: object
     """
+    if cached:
+        return result_cached(task_id, wait)
     start = time.time()
     while True:
         r = Task.get_result(task_id)
@@ -104,7 +109,21 @@ def result(task_id, wait=0):
         time.sleep(0.01)
 
 
-def result_group(group_id, failures=False):
+def result_cached(task_id, wait=0, broker=None):
+    if not broker:
+        broker = get_broker()
+    key = 'django_q:{}:results'.format(broker.list_key)
+    start = time.time()
+    while True:
+        r = broker.cache.get('{}:{}'.format(key, task_id))
+        if r:
+            return signing.SignedPackage.loads(r)['result']
+        if (time.time() - start) * 1000 >= wait:
+            break
+        time.sleep(0.01)
+
+
+def result_group(group_id, failures=False, cached=Conf.CACHED):
     """
     Return a list of results for a task group.
 
@@ -112,10 +131,26 @@ def result_group(group_id, failures=False):
     :param bool failures: set to True to include failures
     :return: list or results
     """
+    if cached:
+        return result_group_cached(group_id, failures)
     return Task.get_result_group(group_id, failures)
 
 
-def fetch(task_id, wait=0):
+def result_group_cached(group_id, failures=False, broker=None):
+    if not broker:
+        broker = get_broker()
+    key = 'django_q:{}:results'.format(broker.list_key)
+    group_list = broker.cache.get('{}:{}'.format(key, group_id))
+    if group_list:
+        result_list = []
+        for task_package in group_list:
+            task = signing.SignedPackage.loads(task_package)
+            if task['success'] or failures:
+                result_list.append(task['result'])
+        return result_list
+
+
+def fetch(task_id, wait=0, cached=Conf.CACHED):
     """
     Return the processed task.
 
@@ -126,6 +161,8 @@ def fetch(task_id, wait=0):
     :return: the full task object
     :rtype: Task
     """
+    if cached:
+        return fetch_cached(task_id, wait)
     start = time.time()
     while True:
         t = Task.get_task(task_id)
@@ -136,7 +173,32 @@ def fetch(task_id, wait=0):
         time.sleep(0.01)
 
 
-def fetch_group(group_id, failures=True):
+def fetch_cached(task_id, wait=0, broker=None):
+    if not broker:
+        broker = get_broker()
+    key = 'django_q:{}:results'.format(broker.list_key)
+    start = time.time()
+    while True:
+        r = broker.cache.get('{}:{}'.format(key, task_id))
+        if r:
+            task = signing.SignedPackage.loads(r)
+            t = Task(id=task['id'],
+                     name=task['name'],
+                     func=task['func'],
+                     hook=task.get('hook'),
+                     args=task['args'],
+                     kwargs=task['kwargs'],
+                     started=task['started'],
+                     stopped=task['stopped'],
+                     result=task['result'],
+                     success=task['success'])
+            return t
+        if (time.time() - start) * 1000 >= wait:
+            break
+        time.sleep(0.01)
+
+
+def fetch_group(group_id, failures=True, cached=True):
     """
     Return a list of Tasks for a task group.
 
@@ -144,10 +206,37 @@ def fetch_group(group_id, failures=True):
     :param bool failures: set to False to exclude failures
     :return: list of Tasks
     """
+    if cached:
+        return fetch_group_cached(group_id, failures)
     return Task.get_task_group(group_id, failures)
 
 
-def count_group(group_id, failures=False):
+def fetch_group_cached(group_id, failures=True, broker=None):
+    if not broker:
+        broker = get_broker()
+    key = 'django_q:{}:results'.format(broker.list_key)
+    group_list = broker.cache.get('{}:{}'.format(key, group_id))
+    if group_list:
+        task_list = []
+        for task_package in group_list:
+            task = signing.SignedPackage.loads(task_package)
+            if task['success'] or failures:
+                t = Task(id=task['id'],
+                         name=task['name'],
+                         func=task['func'],
+                         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'])
+                task_list.append(t)
+        return task_list
+
+
+def count_group(group_id, failures=False, cached=Conf.CACHED):
     """
     Count the results in a group.
 
@@ -156,10 +245,28 @@ def count_group(group_id, failures=False):
     :return: the number of tasks/results in a group
     :rtype: int
     """
+    if cached:
+        return count_group_cached(group_id, failures)
     return Task.get_group_count(group_id, failures)
 
 
-def delete_group(group_id, tasks=False):
+def count_group_cached(group_id, failures=False, broker=None):
+    if not broker:
+        broker = get_broker()
+    key = 'django_q:{}:results'.format(broker.list_key)
+    group_list = broker.cache.get('{}:{}'.format(key, group_id))
+    if group_list:
+        if not failures:
+            return len(group_list)
+        failure_count = 0
+        for task_package in group_list:
+            task = signing.SignedPackage.loads(task_package)
+            if not task['success']:
+                failure_count += 1
+        return failure_count
+
+
+def delete_group(group_id, tasks=False, cached=Conf.CACHE):
     """
     Delete a group.
 
@@ -168,9 +275,18 @@ def delete_group(group_id, tasks=False):
     Otherwise just the group label is removed.
     :return:
     """
+    if cached:
+        return delete_group_cached(group_id)
     return Task.delete_group(group_id, tasks)
 
 
+def delete_group_cached(group_id, broker=None):
+    if not broker:
+        broker = get_broker()
+    key = 'django_q:{}:results'.format(broker.list_key)
+    return broker.cache.delete('{}:{}'.format(key, group_id))
+
+
 def queue_size(broker=None):
     """
     Returns the current queue size.

django_q/tests/test_cached.py

@@ -0,0 +1,52 @@
+from multiprocessing import Event
+
+import pytest
+
+from django_q.cluster import Sentinel
+from django_q.conf import Conf
+from django_q.tasks import async, result, fetch, count_group, result_group, fetch_group, delete_group
+from django_q.brokers import get_broker
+
+
+@pytest.fixture
+def broker():
+    Conf.DISQUE_NODES = None
+    Conf.IRON_MQ = None
+    Conf.SQS = None
+    Conf.ORM = None
+    Conf.MONGO = None
+    Conf.DJANGO_REDIS = 'default'
+    return get_broker()
+
+
+@pytest.mark.django_db
+def test_cached(broker):
+    broker.purge_queue()
+    broker.cache.clear()
+    group = 'cache_test'
+    # queue the tests
+    task_id = async('math.copysign', 1, -1, cached=True, broker=broker)
+    async('math.copysign', 1, -1, cached=True, broker=broker, group=group)
+    async('math.copysign', 1, -1, cached=True, broker=broker, group=group)
+    async('math.copysign', 1, -1, cached=True, broker=broker, group=group)
+    async('math.copysign', 1, -1, cached=True, broker=broker, group=group)
+    async('math.copysign', 1, -1, cached=True, broker=broker, group=group)
+    # run a single cluster
+    start_event = Event()
+    stop_event = Event()
+    stop_event.set()
+    Sentinel(stop_event, start_event, broker=broker)
+    # assert results
+    assert result(task_id, wait=500, cached=True) == -1
+    assert fetch(task_id,wait=500, cached=True).result == -1
+    # make sure it's not in the db backend
+    assert fetch(task_id) is None
+    # assert group
+    assert count_group(group, cached=True) == 5
+    assert count_group(group, cached=True, failures=True) == 0
+    assert result_group(group, cached=True) == [-1, -1, -1, -1, -1]
+    assert len(fetch_group(group, cached=True)) == 5
+    assert len(fetch_group(group, cached=True, failures=False)) == 5
+    delete_group(group, cached=True)
+    assert count_group(group, cached=True) is None
+    broker.cache.clear()