From c127604f1054554310490e113d206a356d247ba0 Mon Sep 17 00:00:00 2001 From: Ilan Date: Sat, 18 Jul 2015 17:20:50 +0200 Subject: [PATCH 1/7] Adds task groups * task now takes an optional keyword `group` * added result_group() and fetch_group() * added group column to Success admin * group column in searchable * added `name` field to Schedules * Scheduled tasks take schedule name as group --- django_q/__init__.py | 2 +- django_q/admin.py | 8 +++-- django_q/cluster.py | 11 +++--- .../migrations/0005_auto_20150718_1506.py | 24 +++++++++++++ django_q/models.py | 13 ++++++- django_q/tasks.py | 36 ++++++++++++++++--- django_q/tests/test_cluster.py | 8 +++-- 7 files changed, 85 insertions(+), 17 deletions(-) create mode 100644 django_q/migrations/0005_auto_20150718_1506.py diff --git a/django_q/__init__.py b/django_q/__init__.py index 7e5a656..88010c5 100644 --- a/django_q/__init__.py +++ b/django_q/__init__.py @@ -4,7 +4,7 @@ import sys myPath = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, myPath) -from .tasks import async, schedule, result, fetch +from .tasks import async, schedule, result, result_group, fetch, fetch_group from .models import Task, Schedule, Success, Failure from .cluster import Cluster diff --git a/django_q/admin.py b/django_q/admin.py index 7dacac5..2f80ade 100644 --- a/django_q/admin.py +++ b/django_q/admin.py @@ -11,7 +11,8 @@ class TaskAdmin(admin.ModelAdmin): 'func', 'started', 'stopped', - 'time_taken' + 'time_taken', + 'group' ) def has_add_permission(self, request, obj=None): @@ -23,7 +24,7 @@ class TaskAdmin(admin.ModelAdmin): qs = super(TaskAdmin, self).get_queryset(request) return qs.filter(success=True) - search_fields = ('name', 'func') + search_fields = ('name', 'func', 'group') readonly_fields = [] def get_readonly_fields(self, request, obj=None): @@ -65,6 +66,7 @@ class FailAdmin(admin.ModelAdmin): class ScheduleAdmin(admin.ModelAdmin): list_display = ( 'id', + 'name', 'func', 'schedule_type', 'repeats', @@ -75,7 +77,7 @@ class ScheduleAdmin(admin.ModelAdmin): list_filter = ('next_run', 'schedule_type') search_fields = ('func',) - list_display_links = ('id', 'func') + list_display_links = ('id', 'name') admin.site.register(Schedule, ScheduleAdmin) diff --git a/django_q/cluster.py b/django_q/cluster.py index ad2edab..6ad88c2 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -404,12 +404,13 @@ def save_task(task): Task.objects.create(id=task['id'], name=task['name'], func=task['func'], - hook=task['hook'], + 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']) except Exception as e: logger.error(e) @@ -455,13 +456,15 @@ def scheduler(list_key=Conf.Q_LIST): s.repeats += -1 # send it to the cluster kwargs['list_key'] = list_key + kwargs['group'] = s.name or s.id s.task = tasks.async(s.func, *args, **kwargs) # log it if not s.task: - logger.error(_('{} failed to create a task from schedule {} [{}]').format(current_process().name, s.id), - s.func) + logger.error( + _('{} failed to create a task from schedule [{}]').format(current_process().name, s.name or s.id)) else: - logger.info(_('{} created a task from schedule {} [{}]').format(current_process().name, s.id, s.func)) + logger.info( + _('{} created a task from schedule [{}]').format(current_process().name, s.name or s.id)) # default behavior is to delete a ONCE schedule if s.schedule_type == s.ONCE: if s.repeats < 0: diff --git a/django_q/migrations/0005_auto_20150718_1506.py b/django_q/migrations/0005_auto_20150718_1506.py new file mode 100644 index 0000000..e0636c6 --- /dev/null +++ b/django_q/migrations/0005_auto_20150718_1506.py @@ -0,0 +1,24 @@ +# -*- coding: utf-8 -*- +from __future__ import unicode_literals + +from django.db import models, migrations + + +class Migration(migrations.Migration): + + dependencies = [ + ('django_q', '0004_auto_20150710_1043'), + ] + + operations = [ + migrations.AddField( + model_name='schedule', + name='name', + field=models.CharField(max_length=256, null=True), + ), + migrations.AddField( + model_name='task', + name='group', + field=models.CharField(max_length=100, null=True, editable=False), + ), + ] diff --git a/django_q/models.py b/django_q/models.py index 98b3b62..21e6ec1 100644 --- a/django_q/models.py +++ b/django_q/models.py @@ -1,6 +1,6 @@ -import importlib import logging +import importlib from django.core.urlresolvers import reverse from django.utils.translation import ugettext_lazy as _ from django.db import models @@ -18,6 +18,7 @@ class Task(models.Model): args = PickledObjectField(null=True) kwargs = PickledObjectField(null=True) result = PickledObjectField(null=True) + group = models.CharField(max_length=100, editable=False, null=True) started = models.DateTimeField(editable=False) stopped = models.DateTimeField(editable=False) success = models.BooleanField(default=True, editable=False) @@ -29,6 +30,11 @@ class Task(models.Model): elif Task.objects.filter(name=task_id).exists(): return Task.objects.get(name=task_id).result + @staticmethod + def get_group_result(group_id): + # values_list() doesn't work here cause it returns encoded fields + return [t.result for t in Task.get_task_group(group_id)] + @staticmethod def get_task(task_id): if len(task_id) == 32 and Task.objects.filter(id=task_id).exists(): @@ -36,6 +42,10 @@ class Task(models.Model): elif Task.objects.filter(name=task_id).exists(): return Task.objects.get(name=task_id) + @staticmethod + def get_task_group(group_id): + return Task.objects.filter(group=group_id) + def time_taken(self): return (self.stopped - self.started).total_seconds() @@ -101,6 +111,7 @@ class Failure(Task): class Schedule(models.Model): + name = models.CharField(max_length=256, null=True) func = models.CharField(max_length=256, help_text='e.g. module.tasks.function') hook = models.CharField(max_length=256, null=True, blank=True, help_text='e.g. module.tasks.result_function') args = models.TextField(null=True, blank=True, help_text=_("e.g. 1, 2, 'John'")) diff --git a/django_q/tasks.py b/django_q/tasks.py index 9747fd1..9cc3ef6 100644 --- a/django_q/tasks.py +++ b/django_q/tasks.py @@ -25,20 +25,27 @@ def async(func, *args, **kwargs): # optional list_key list_key = kwargs.pop('list_key', Conf.Q_LIST) # optional redis connection - r = kwargs.pop('redis', redis_client) + redis = kwargs.pop('redis', redis_client) # optional sync mode - s = kwargs.pop('sync', False) + sync = kwargs.pop('sync', False) + # optional group + group = kwargs.pop('group', None) # get an id tag = uuid() # build the task package - task = {'id': tag[1], 'name': tag[0], 'func': func, 'hook': hook, 'args': args, 'kwargs': kwargs, + task = {'id': tag[1], 'name': tag[0], 'func': func, 'args': args, 'kwargs': kwargs, 'started': timezone.now()} + # add optionals + if hook: + task['hook'] = hook + if group: + task['group'] = group # sign it pack = signing.SignedPackage.dumps(task) - if s: + if sync: return _sync(task['id'], pack) # push it - r.rpush(list_key, pack) + redis.rpush(list_key, pack) logger.debug('Pushed {}'.format(tag)) return task['id'] @@ -83,6 +90,15 @@ def result(task_id): return Task.get_result(task_id) +def result_group(group_id): + """ + returns a list of results for a task group + :param str group_id: the group id + :return: list or results + """ + return Task.get_group_result(group_id) + + def fetch(task_id): """ Returns the processed task @@ -94,6 +110,16 @@ def fetch(task_id): return Task.get_task(task_id) +def fetch_group(group_id): + """ + Returns a list of Tasks for a task group + :param str group_id: the group id + :return: list of Tasks + """ + + return Task.get_task_group(group_id) + + def _sync(task_id, pack): """ Simulates a package travelling through the cluster. diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py index 32fba17..c51ea08 100644 --- a/django_q/tests/test_cluster.py +++ b/django_q/tests/test_cluster.py @@ -10,7 +10,7 @@ sys.path.insert(0, myPath + '/../') from django_q.cluster import Cluster, Sentinel, pusher, worker, monitor from django_q.humanhash import DEFAULT_WORDLIST -from django_q.tasks import fetch, async, result +from django_q.tasks import fetch, fetch_group, async, result, result_group from django_q.models import Task from django_q.conf import Conf, redis_client from .tasks import multiply @@ -119,10 +119,10 @@ def test_async(r, admin_user): f = async(multiply, 753, 2, hook=assert_result, list_key=list_key, redis=r) # model as argument g = async('django_q.tests.tasks.get_task_name', Task(name='John'), list_key=list_key, redis=r) - # args and kwargs and broken hook + # args,kwargs, group and broken hook 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, redis=r) + j = async('django_q.tests.tasks.get_user_id', admin_user, list_key=list_key, group='test_j', redis=r) # check if everything has a task id assert isinstance(a, str) assert isinstance(b, str) @@ -197,6 +197,8 @@ def test_async(r, admin_user): assert result_j is not None assert result_j.success is True assert result_j.result == result_j.args[0].id + assert result_group('test_j') == [result_j.result] + assert fetch_group('test_j')[0].id == [result_j][0].id r.delete(list_key) From a5d4c5186c8b7c488da41f15fef4c0a27dfbe64f Mon Sep 17 00:00:00 2001 From: Ilan Date: Sat, 18 Jul 2015 18:49:50 +0200 Subject: [PATCH 2/7] renamed `get_group_result` to `get_result_group` --- django_q/tasks.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/django_q/tasks.py b/django_q/tasks.py index 9cc3ef6..f952f3d 100644 --- a/django_q/tasks.py +++ b/django_q/tasks.py @@ -96,7 +96,7 @@ def result_group(group_id): :param str group_id: the group id :return: list or results """ - return Task.get_group_result(group_id) + return Task.get_result_group(group_id) def fetch(task_id): From e1d483e1c14beb6518840aa66628243fb84f4fbc Mon Sep 17 00:00:00 2001 From: Ilan Date: Sat, 18 Jul 2015 18:50:14 +0200 Subject: [PATCH 3/7] docs: describes the new group options --- docs/schedules.rst | 6 +++- docs/tasks.rst | 71 ++++++++++++++++++++++++++++++++++++++++------ 2 files changed, 67 insertions(+), 10 deletions(-) diff --git a/docs/schedules.rst b/docs/schedules.rst index b2c1780..a5dbe2a 100644 --- a/docs/schedules.rst +++ b/docs/schedules.rst @@ -51,7 +51,7 @@ Reference .. py:function:: schedule(func, *args, hook=None, schedule_type='O', repeats=-1, next_run=now() , **kwargs) Creates a schedule - + :param str name: A sensible name for your schedule :param str func: the function to schedule. Dotted strings only. :param args: arguments for the scheduled function. :param str hook: optional result hook function. Dotted strings only. @@ -68,6 +68,10 @@ Reference Primary key + .. py:attribute:: name + + A name for your schedule. Tasks created by this schedule will assume this as their group id. + .. py:attribute:: func The function to be scheduled diff --git a/docs/tasks.rst b/docs/tasks.rst index c3fd630..bff8ef7 100644 --- a/docs/tasks.rst +++ b/docs/tasks.rst @@ -31,6 +31,31 @@ Use :func:`async` from your code to quickly offload tasks to the :class:`Cluster def print_result(task): print(task.result) +Groups +------ +You can group together results by passing :func:`async` the optional `group` keyword: + +.. code-block:: python + + from django_q import async, result_group + + for i in range(4): + async('math.modf', i, group='modf') + + # after the tasks have finished you can get the group results + result = result_group('modf') + print(result) + +.. code-block:: python + + [(0.0, 0.0), (0.0, 1.0), (0.0, 2.0), (0.0, 3.0)] + +Take care that you haven't limited your results database too much and that the group identifier is unique for each run. +Instead of :func:`result_group` you can also use :func:`fetch_group` to return a list of :class:`Task` objects. + +.. note:: + Tasks created by a schedule, will use the schedules name as their group id. + Synchronous testing ------------------- @@ -78,19 +103,19 @@ When you are making individual calls to :func:`async` a lot though, it can help Reference --------- -.. py:function:: async(func, *args, hook=None, timeout=None, sync=False, redis=None, **kwargs) +.. py:function:: async(func, *args, hook=None, group=None, timeout=None,\ + sync=False, redis=None, **kwargs) Puts a task in the cluster queue - :param func: The task function to execute - :param args: The arguments for the task function - :type func: object - :param hook: Optional function to call after execution - :type hook: object + :param object func: The task function to execute + :param tuple args: The arguments for the task function + :param object hook: Optional function to call after execution + :param str group: An optional group identifier :param int timeout: Overrides global cluster :ref:`timeout`. :param bool sync: If set to True, async will simulate a task execution :param redis: Optional redis connection - :param kwargs: Keyword arguments for the task function + :param dict kwargs: Keyword arguments for the task function :returns: The uuid of the task :rtype: str @@ -106,13 +131,29 @@ Reference Returns a previously executed task :param str name: the uuid or name of the task - :returns: The task + :returns: The task if any :rtype: Task .. versionchanged:: 0.2.0 Renamed from get_task +.. py:function:: result_group(group_id) + + Returns the results of a task group + + :param str group_id: the group identifier + :returns: a list of results + :rtype: list + +.. py:function:: fetch_group(group_id) + + Returns a list of tasks in a group + + :param str group_id: the group identifier + :returns: a list of Tasks + :rtype: list + .. py:class:: Task Database model describing an executed task @@ -174,7 +215,19 @@ Reference .. py:classmethod:: get_result(task_id) - Get a result directly by task uuid or name + Gets a result directly by task uuid or name. + + .. py:classmethod:: get_result_group(group_id) + + Returns a list of results from a task group. + + .. py:classmethod:: get_task(task_id) + + Fetches a single task object by uuid or name. + + .. py:classmethod:: get_task_group(group_id) + + Gets a queryset of tasks with this group id. .. py:class:: Success From 05d6ecb7a2be3e023a7cd33df175d730371d4d0b Mon Sep 17 00:00:00 2001 From: Ilan Date: Sat, 18 Jul 2015 18:55:47 +0200 Subject: [PATCH 4/7] renamed get_group_result to get_result_group I think I fumbled a commit --- django_q/models.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/django_q/models.py b/django_q/models.py index 21e6ec1..2294cc0 100644 --- a/django_q/models.py +++ b/django_q/models.py @@ -31,7 +31,7 @@ class Task(models.Model): return Task.objects.get(name=task_id).result @staticmethod - def get_group_result(group_id): + def get_result_group(group_id): # values_list() doesn't work here cause it returns encoded fields return [t.result for t in Task.get_task_group(group_id)] From e4737d765a16e369f4582eacb6dc5921aed28de6 Mon Sep 17 00:00:00 2001 From: Ilan Date: Sat, 18 Jul 2015 19:41:21 +0200 Subject: [PATCH 5/7] Schedule name is optional For backwards compatibility the schedule name is an optional keyword in the `schedule` function. Adjusted docs accordingly --- django_q/tasks.py | 5 ++++- django_q/tests/test_scheduler.py | 1 + docs/admin.rst | 3 ++- docs/schedules.rst | 7 ++++--- docs/tasks.rst | 2 +- 5 files changed, 12 insertions(+), 6 deletions(-) diff --git a/django_q/tasks.py b/django_q/tasks.py index f952f3d..069294c 100644 --- a/django_q/tasks.py +++ b/django_q/tasks.py @@ -54,6 +54,7 @@ def schedule(func, *args, **kwargs): """ :param func: function to schedule :param args: function arguments + :param name: optional name for the schedule :param hook: optional result hook function :type schedule_type: Schedule.TYPE :param repeats: how many times to repeat. 0=never, -1=always @@ -64,12 +65,14 @@ def schedule(func, *args, **kwargs): :rtype: Schedule """ + name = kwargs.pop('name', None) 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, + return Schedule.objects.create(name=name, + func=func, hook=hook, args=args, kwargs=kwargs, diff --git a/django_q/tests/test_scheduler.py b/django_q/tests/test_scheduler.py index c6860d1..8fc3505 100644 --- a/django_q/tests/test_scheduler.py +++ b/django_q/tests/test_scheduler.py @@ -20,6 +20,7 @@ def test_scheduler(r): r.delete(list_key) schedule = create_schedule('math.copysign', 1, -1, + name='test math', hook='django_q.tests.tasks.result', schedule_type=Schedule.HOURLY, repeats=1) diff --git a/docs/admin.rst b/docs/admin.rst index f1c3fd4..b4c57a4 100644 --- a/docs/admin.rst +++ b/docs/admin.rst @@ -11,7 +11,8 @@ Successful tasks ---------------- Shows all successfully executed tasks. Meaning they did not encounter any errors during execution. -From here you can look at details of each task or delete them. +From here you can look at details of each task or delete them. Use the group column to sort your results by schedule name or group id. +The table is searchable by `name`, `func` and `group` Uses the :class:`Success` proxy model. diff --git a/docs/schedules.rst b/docs/schedules.rst index a5dbe2a..c726e0d 100644 --- a/docs/schedules.rst +++ b/docs/schedules.rst @@ -48,12 +48,13 @@ If you want to schedule regular Django management commands, you can use the :mod Reference --------- -.. py:function:: schedule(func, *args, hook=None, schedule_type='O', repeats=-1, next_run=now() , **kwargs) +.. py:function:: schedule(func, *args, name=None, hook=None, schedule_type='O', repeats=-1, next_run=now() , **kwargs) Creates a schedule - :param str name: A sensible name for your schedule + :param str func: the function to schedule. Dotted strings only. :param args: arguments for the scheduled function. + :param str name: An optional name for your schedule. :param str hook: optional result hook function. Dotted strings only. :param str schedule_type: (O)nce, (H)ourly, (D)aily, (W)eekly, (M)onthly, (Q)uarterly, (Y)early or :attr:`Schedule.TYPE` :param int repeats: Number of times to repeat schedule. -1=Always, 0=Never, n =n. @@ -70,7 +71,7 @@ Reference .. py:attribute:: name - A name for your schedule. Tasks created by this schedule will assume this as their group id. + A name for your schedule. Tasks created by this schedule will assume this or the primary key as their group id. .. py:attribute:: func diff --git a/docs/tasks.rst b/docs/tasks.rst index bff8ef7..862be40 100644 --- a/docs/tasks.rst +++ b/docs/tasks.rst @@ -54,7 +54,7 @@ Take care that you haven't limited your results database too much and that the g Instead of :func:`result_group` you can also use :func:`fetch_group` to return a list of :class:`Task` objects. .. note:: - Tasks created by a schedule, will use the schedules name as their group id. + Tasks created by a schedule, will use the schedule name as their group id. This enables you to browse the schedule's history. Synchronous testing ------------------- From 58a671fb455d0532ec17039f25e1e19c4d305ee3 Mon Sep 17 00:00:00 2001 From: Ilan Date: Sat, 18 Jul 2015 20:50:16 +0200 Subject: [PATCH 6/7] docs: added a `fetch_group` example --- docs/tasks.rst | 22 +++++++++++++++++++--- 1 file changed, 19 insertions(+), 3 deletions(-) diff --git a/docs/tasks.rst b/docs/tasks.rst index 862be40..656b2fc 100644 --- a/docs/tasks.rst +++ b/docs/tasks.rst @@ -37,6 +37,7 @@ You can group together results by passing :func:`async` the optional `group` key .. code-block:: python + # result group example from django_q import async, result_group for i in range(4): @@ -50,11 +51,26 @@ You can group together results by passing :func:`async` the optional `group` key [(0.0, 0.0), (0.0, 1.0), (0.0, 2.0), (0.0, 3.0)] -Take care that you haven't limited your results database too much and that the group identifier is unique for each run. -Instead of :func:`result_group` you can also use :func:`fetch_group` to return a list of :class:`Task` objects. +Take care to not limit your results database too much and that the group identifier is unique for each run. +Instead of :func:`result_group` you can also use :func:`fetch_group` to return a queryset of :class:`Task` objects.: + +.. code-block:: python + + # fetch group example + from django_q import fetch_group + + # count the number of failures + failure_count = fetch_group('modf').filter(success=False).count() + + # or print only the successful results + successes = fetch_group('modf').exclude(success=False) + results = [task.result for task in successes] + print(results) .. note:: - Tasks created by a schedule, will use the schedule name as their group id. This enables you to browse the schedule's history. + + Although :func:`fetch_group` returns a queryset, due to the nature of the PickleField , `Queryset.values` will return a list of encoded results. + Use list comprehension or an iterator instead. Synchronous testing ------------------- From f65c05624d72d059f15057a2f59d2e9a19a6cded Mon Sep 17 00:00:00 2001 From: Ilan Date: Sat, 18 Jul 2015 22:32:32 +0200 Subject: [PATCH 7/7] Schedule name and task group should have same max length --- django_q/migrations/0005_auto_20150718_1506.py | 2 +- django_q/models.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/django_q/migrations/0005_auto_20150718_1506.py b/django_q/migrations/0005_auto_20150718_1506.py index e0636c6..fdb36ca 100644 --- a/django_q/migrations/0005_auto_20150718_1506.py +++ b/django_q/migrations/0005_auto_20150718_1506.py @@ -14,7 +14,7 @@ class Migration(migrations.Migration): migrations.AddField( model_name='schedule', name='name', - field=models.CharField(max_length=256, null=True), + field=models.CharField(max_length=100, null=True), ), migrations.AddField( model_name='task', diff --git a/django_q/models.py b/django_q/models.py index 2294cc0..78b4ed4 100644 --- a/django_q/models.py +++ b/django_q/models.py @@ -111,7 +111,7 @@ class Failure(Task): class Schedule(models.Model): - name = models.CharField(max_length=256, null=True) + name = models.CharField(max_length=100, null=True) func = models.CharField(max_length=256, help_text='e.g. module.tasks.function') hook = models.CharField(max_length=256, null=True, blank=True, help_text='e.g. module.tasks.result_function') args = models.TextField(null=True, blank=True, help_text=_("e.g. 1, 2, 'John'"))