diff --git a/django_q/__init__.py b/django_q/__init__.py index e622c82..10c9635 100644 --- a/django_q/__init__.py +++ b/django_q/__init__.py @@ -1,6 +1,6 @@ from .tasks import async, schedule, result, fetch from .models import Task, Schedule -VERSION = (0, 2, 2) +VERSION = (0, 3, 0) default_app_config = 'django_q.apps.DjangoQConfig' diff --git a/django_q/cluster.py b/django_q/cluster.py index e2e6f5b..6e1f233 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -394,7 +394,8 @@ def save_task(task): Success.objects.first().delete() try: - Task.objects.create(name=task['name'], + Task.objects.create(id=task['id'], + name=task['name'], func=task['func'], hook=task['hook'], args=task['args'], @@ -451,7 +452,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 a task from schedule {} [{}]').format(current_process().name, s.id), s.func) else: - logger.info(_('{} created [{}] from schedule {}').format(current_process().name, s.task, s.id)) + logger.info(_('{} created a task from schedule {} [{}]').format(current_process().name, s.id, s.func)) s.save() diff --git a/django_q/migrations/0003_auto_20150708_1326.py b/django_q/migrations/0003_auto_20150708_1326.py new file mode 100644 index 0000000..05ad1b0 --- /dev/null +++ b/django_q/migrations/0003_auto_20150708_1326.py @@ -0,0 +1,31 @@ +# -*- coding: utf-8 -*- +from __future__ import unicode_literals + +from django.db import models, migrations + + +class Migration(migrations.Migration): + + dependencies = [ + ('django_q', '0002_auto_20150630_1624'), + ] + + operations = [ + migrations.AlterModelOptions( + name='failure', + options={'verbose_name_plural': 'Failed tasks', 'verbose_name': 'Failed task'}, + ), + migrations.AlterModelOptions( + name='schedule', + options={'verbose_name_plural': 'Scheduled tasks', 'ordering': ['next_run'], 'verbose_name': 'Scheduled task'}, + ), + migrations.AlterModelOptions( + name='success', + options={'verbose_name_plural': 'Successful tasks', 'verbose_name': 'Successful task'}, + ), + migrations.AlterField( + model_name='task', + name='id', + field=models.CharField(max_length=32, primary_key=True, editable=False, serialize=False), + ), + ] diff --git a/django_q/models.py b/django_q/models.py index 1e00d68..d16432a 100644 --- a/django_q/models.py +++ b/django_q/models.py @@ -11,6 +11,7 @@ from picklefield import PickledObjectField class Task(models.Model): + id = models.CharField(max_length=32, primary_key=True, editable=False) name = models.CharField(max_length=100, editable=False) func = models.CharField(max_length=256) hook = models.CharField(max_length=256, null=True) @@ -22,9 +23,18 @@ class Task(models.Model): success = models.BooleanField(default=True, editable=False) @staticmethod - def get_result(name): - if Task.objects.filter(name=name).exists(): - return Task.objects.get(name=name).result + def get_result(task_id): + if len(task_id) == 32 and Task.objects.filter(id=task_id).exists(): + return Task.objects.get(id=task_id).result + elif Task.objects.filter(name=task_id).exists(): + return Task.objects.get(name=task_id).result + + @staticmethod + def get_task(task_id): + if len(task_id) == 32 and Task.objects.filter(id=task_id).exists(): + return Task.objects.get(id=task_id) + elif Task.objects.filter(name=task_id).exists(): + return Task.objects.get(name=task_id) def time_taken(self): return (self.stopped - self.started).total_seconds() @@ -47,12 +57,12 @@ def call_hook(sender, instance, **kwargs): m = importlib.import_module(module) f = getattr(m, func) except (ValueError, ImportError, AttributeError): - logger.error(_('malformed return hook \'{}\' for {}').format(instance.hook, instance.name)) + logger.error(_('malformed return hook \'{}\' for [{}]').format(instance.hook, instance.name)) return try: f(instance) except Exception as e: - logger.error(_('return hook {} failed on {} because {}').format(instance.hook, instance.name, e)) + logger.error(_('return hook {} failed on [{}] because {}').format(instance.hook, instance.name, e)) class SuccessManager(models.Manager): @@ -111,28 +121,26 @@ class Schedule(models.Model): schedule_type = models.CharField(max_length=1, choices=TYPE, default=TYPE[0][0], verbose_name=_('Schedule Type')) repeats = models.SmallIntegerField(default=-1, verbose_name=_('Repeats'), help_text=_('n = n times, -1 = forever')) next_run = models.DateTimeField(verbose_name=_('Next Run'), default=timezone.now, null=True) - task = models.CharField(max_length=100, editable=False, null=True) + task = models.CharField(max_length=100, null=True, editable=False) + + def success(self): + if self.task and Task.objects.filter(id=self.task): + return Task.objects.get(id=self.task).success def last_run(self): - if Task.objects.filter(name=self.task).exists(): - task = Task.objects.get(name=self.task) + if self.task and Task.objects.filter(id=self.task): + task = Task.objects.get(id=self.task) if task.success: url = reverse('admin:django_q_success_change', args=(task.id,)) else: url = reverse('admin:django_q_failure_change', args=(task.id,)) return '[{}]'.format(url, self.task) - return None - def success(self): - if Task.objects.filter(name=self.task).exists(): - return Task.objects.get(name=self.task).success - def __unicode__(self): return self.func success.boolean = True - last_run.allow_tags = True class Meta: app_label = 'django_q' diff --git a/django_q/monitor.py b/django_q/monitor.py index e946f61..2ae93d4 100644 --- a/django_q/monitor.py +++ b/django_q/monitor.py @@ -43,6 +43,7 @@ def monitor(run_once=False): stats = Stat.get_all(r=r) print(term.clear_eos()) for stat in stats: + status = stat.status # color status if stat.status == Conf.WORKING: status = term.green(str(Conf.WORKING)) diff --git a/django_q/tasks.py b/django_q/tasks.py index 17f7235..3bf9478 100644 --- a/django_q/tasks.py +++ b/django_q/tasks.py @@ -13,8 +13,6 @@ from .models import Schedule, Task from .humanhash import uuid - - def async(func, *args, **kwargs): """ Sends a task to the cluster @@ -23,12 +21,13 @@ def async(func, *args, **kwargs): 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()} + tag = uuid() + task = {'id': tag[1], 'name': tag[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'] + logger.debug('Pushed {}'.format(tag)) + return task['id'] def schedule(func, *args, **kwargs): @@ -60,27 +59,26 @@ def schedule(func, *args, **kwargs): ) -def result(name): +def result(task_id): """ Returns the result of the named task - :type name: str or unicode - :param name: the task name + :type task_id: str or uuid + :param task_id: the task name or uuid :return: the result object of this task - :rtype: object or str + :rtype: object """ - return Task.get_result(name) + return Task.get_result(task_id) -def fetch(name): +def fetch(task_id): """ Returns the processed task - :param name: the task name - :type name: str or unicode + :param task_id: the task name or uuid + :type task_id: str or uuid :return: the full task object :rtype: Task """ - if Task.objects.filter(name=name).exists(): - return Task.objects.get(name=name) + return Task.get_task(task_id) class SignedPackage(object): @@ -116,4 +114,4 @@ class PickleSerializer(object): @staticmethod def loads(data): - return pickle.loads(data) \ No newline at end of file + return pickle.loads(data) diff --git a/django_q/tests/test_admin.py b/django_q/tests/test_admin.py index 9e2b525..a18bcdf 100644 --- a/django_q/tests/test_admin.py +++ b/django_q/tests/test_admin.py @@ -1,26 +1,32 @@ from django.core.urlresolvers import reverse from django.utils import timezone - import pytest from django_q.tasks import schedule from django_q.models import Task, Failure +from django_q.humanhash import uuid @pytest.mark.django_db def test_admin_views(admin_client): s = schedule('sched.test') - f = Task.objects.create(name='alfa-pappa-bravo-fail', - func='test.fail', - started=timezone.now(), - stopped=timezone.now(), - success=False) - t = Task.objects.create(name='alfa-pappa-bravo-success', - func='test.succes', - started=timezone.now(), - stopped=timezone.now(), - success=True) + tag = uuid() + f = Task.objects.create( + id=tag[1], + name=tag[0], + func='test.fail', + started=timezone.now(), + stopped=timezone.now(), + success=False) + tag = uuid() + t = Task.objects.create( + id=tag[1], + name=tag[0], + func='test.succes', + started=timezone.now(), + stopped=timezone.now(), + success=True) admin_urls = ( # schedule reverse('admin:django_q_schedule_changelist'), diff --git a/docs/conf.py b/docs/conf.py index 1a07031..c1583b5 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -68,9 +68,9 @@ author = 'Ilan Steemers' # built documents. # # The short X.Y version. -version = '0.2.0' +version = '0.3' # The full version, including alpha/beta/rc tags. -release = '0.2.0' +release = '0.3.0' # The language for content autogenerated by Sphinx. Refer to documentation # for a list of supported languages. diff --git a/docs/schedules.rst b/docs/schedules.rst index c17c99e..99a8595 100644 --- a/docs/schedules.rst +++ b/docs/schedules.rst @@ -77,7 +77,7 @@ Reference .. py:attribute:: task - Name of the last task generated by this schedule. + Id of the last task generated by this schedule. .. py:method:: last_run() diff --git a/docs/tasks.rst b/docs/tasks.rst index 4a49b5b..e773919 100644 --- a/docs/tasks.rst +++ b/docs/tasks.rst @@ -61,21 +61,21 @@ Reference :type hook: str or object :param redis: Optional redis connection :param kwargs: Keyword arguments for the task function - :returns: The name of the task + :returns: The uuid of the task :rtype: str -.. py:function:: result(name) +.. py:function:: result(task_id) Gets the result of a previously executed task - :param str name: the name of the task + :param str task_id: the uuid or name of the task :returns: The result of the executed task -.. py:function:: fetch(name) +.. py:function:: fetch(task_id) Returns a previously executed task - :param str name: the name of the task + :param str name: the uuid or name of the task :returns: The task :rtype: Task @@ -87,9 +87,18 @@ Reference Database model describing an executed task + .. py:attribute:: id + + An :func:`uuid.uuid4()` identifier + .. py:attribute:: name - The name of the task + The name of the task as a humanized version of the :attr:`id` + + .. note:: + + This is for convenience and can be used as a parameter for most functions that take a `task_id`. + Keep in mind however that it is not guaranteed to be unique if you store very large amounts of tasks in the database. .. py:attribute:: func @@ -115,7 +124,7 @@ Reference .. py:attribute:: started - The moment the task was picked up by a worker + The moment the task was created by an async command .. py:attribute:: stopped @@ -127,11 +136,15 @@ Reference .. py:method:: time_taken - Calculates the difference in seconds between started and stopped + Calculates the difference in seconds between started and stopped. - .. py:classmethod:: get_result(task_name) + .. note:: - Get a result directly by task name + Time taken represents the time a task spends in the cluster, this includes any time it may have waited in the queue. + + .. py:classmethod:: get_result(task_id) + + Get a result directly by task uuid or name .. py:class:: Success diff --git a/setup.py b/setup.py index a65d299..f24280b 100644 --- a/setup.py +++ b/setup.py @@ -26,12 +26,13 @@ class PyTest(Command): setup( name='django-q', - version='0.2.2', + version='0.3.0', author='Ilan Steemers', author_email='koed00@gmail.com', + keywords='django task queue worker redis multiprocessing', packages=['django_q'], include_package_data=True, - url='https://github.com/koed00/django-q', + url='https://django-q.readthedocs.org', license='MIT', description='A multiprocessing task queue for Django', long_description=README,