From c5b8b800f626819e53b4a09d1a10339b770a5eed Mon Sep 17 00:00:00 2001 From: Ilan Date: Mon, 20 Jul 2015 19:43:27 +0200 Subject: [PATCH] Adds `save` override option for tasks Also adds `q_options` dict for async and schedule options --- django_q/cluster.py | 11 ++++++----- django_q/tasks.py | 20 ++++++++++---------- django_q/tests/test_cluster.py | 9 +++++++-- 3 files changed, 23 insertions(+), 17 deletions(-) diff --git a/django_q/cluster.py b/django_q/cluster.py index 6ad88c2..c474c43 100644 --- a/django_q/cluster.py +++ b/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 - kwargs['group'] = s.name or s.id + q_options['list_key'] = list_key + 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: diff --git a/django_q/tasks.py b/django_q/tasks.py index 74dd52b..32cee05 100644 --- a/django_q/tasks.py +++ b/django_q/tasks.py @@ -20,16 +20,14 @@ def async(func, *args, **kwargs): """ Sends a task to the cluster """ - # optional hook - hook = kwargs.pop('hook', None) - # optional list_key - list_key = kwargs.pop('list_key', Conf.Q_LIST) - # optional redis connection - redis = kwargs.pop('redis', redis_client) - # optional sync mode - sync = kwargs.pop('sync', False) - # optional group - group = kwargs.pop('group', None) + # get options from q_options dict or direct from kwargs + options = kwargs.pop('q_options', kwargs) + hook = options.pop('hook', None) + list_key = options.pop('list_key', Conf.Q_LIST) + redis = options.pop('redis', redis_client) + sync = options.pop('sync', False) + group = options.pop('group', None) + save = options.pop('save', 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: diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py index 3c7a858..32a5291 100644 --- a/django_q/tests/test_cluster.py +++ b/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)