Adds save override option for tasks

Also adds `q_options` dict for async and schedule options

django_q/cluster.py

@@ -394,12 +394,11 @@ def save_task(task):
     Saves the task package to Django
     """
     # SAVE LIMIT < 0 : Don't save success
-    if Conf.SAVE_LIMIT < 0 and task['success']:
+    if not task.get('save', Conf.SAVE_LIMIT > 0) and task['success']:
         return
     # SAVE LIMIT > 0: Prune database, SAVE_LIMIT 0: No pruning
     if task['success'] and 0 < Conf.SAVE_LIMIT < Success.objects.count():
         Success.objects.last().delete()
-
     try:
         Task.objects.create(id=task['id'],
                             name=task['name'],
@@ -435,8 +434,9 @@ def scheduler(list_key=Conf.Q_LIST):
             # single value won't eval to tuple, so:
             if type(args) != tuple:
                 args = (args,)
+        q_options = kwargs.get('q_options', {})
         if s.hook:
-            kwargs['hook'] = s.hook
+            q_options['hook'] = s.hook
         # set up the next run time
         if not s.schedule_type == s.ONCE:
             next_run = arrow.get(s.next_run)
@@ -455,8 +455,9 @@ def scheduler(list_key=Conf.Q_LIST):
             s.next_run = next_run.datetime
             s.repeats += -1
         # send it to the cluster
-        kwargs['list_key'] = list_key
+        q_options['list_key'] = list_key
-        kwargs['group'] = s.name or s.id
+        q_options['group'] = s.name or s.id
+        kwargs['q_options'] = q_options
         s.task = tasks.async(s.func, *args, **kwargs)
         # log it
         if not s.task:

django_q/tasks.py

@@ -20,16 +20,14 @@ def async(func, *args, **kwargs):
     """
     Sends a task to the cluster
     """
-    # optional hook
+    # get options from q_options dict or direct from kwargs
-    hook = kwargs.pop('hook', None)
+    options = kwargs.pop('q_options', kwargs)
-    # optional list_key
+    hook = options.pop('hook', None)
-    list_key = kwargs.pop('list_key', Conf.Q_LIST)
+    list_key = options.pop('list_key', Conf.Q_LIST)
-    # optional redis connection
+    redis = options.pop('redis', redis_client)
-    redis = kwargs.pop('redis', redis_client)
+    sync = options.pop('sync', False)
-    # optional sync mode
+    group = options.pop('group', None)
-    sync = kwargs.pop('sync', False)
+    save = options.pop('save', None)
-    # optional group
-    group = kwargs.pop('group', None)
     # get an id
     tag = uuid()
     # build the task package
@@ -40,6 +38,8 @@ def async(func, *args, **kwargs):
         task['hook'] = hook
     if group:
         task['group'] = group
+    if save is not None:
+        task['save'] = save
     # sign it
     pack = signing.SignedPackage.dumps(task)
     if sync:

django_q/tests/test_cluster.py

@@ -124,6 +124,9 @@ def test_async(r, admin_user):
     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, group='test_j', redis=r)
+    # q_options and save opt_out test
+    k = async('django_q.tests.tasks.get_user_id', admin_user,
+              q_options={'list_key': list_key, 'group': 'test_k', 'redis': r, 'save': False, 'timeout': 90})
     # check if everything has a task id
     assert isinstance(a, str)
     assert isinstance(b, str)
@@ -134,8 +137,9 @@ def test_async(r, admin_user):
     assert isinstance(g, str)
     assert isinstance(h, str)
     assert isinstance(j, str)
+    assert isinstance(k, str)
     # run the cluster to execute the tasks
-    task_count = 9
+    task_count = 10
     assert r.llen(list_key) == task_count
     task_queue = Queue()
     stop_event = Event()
@@ -210,7 +214,8 @@ def test_async(r, admin_user):
     assert count_group('test_j', failures=True) == 0
     assert delete_group('test_j') == 1
     assert delete_group('test_j', tasks=True) is None
-
+    # task k should not have been saved
+    assert fetch(k) is None
     r.delete(list_key)