Task now uses a UUID4 instead of a LUID

This commit is contained in:
Ilan Steemers
2015-07-08 13:33:03 +02:00
parent ddf861a16d
commit f8708cfd73
8 changed files with 91 additions and 49 deletions
+4 -3
View File
@@ -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()
@@ -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', '0001_squashed_0003_auto_20150707_1854'),
]
operations = [
migrations.AlterField(
model_name='schedule',
name='task',
field=models.CharField(null=True, max_length=32, editable=False),
),
migrations.AlterField(
model_name='task',
name='id',
field=models.CharField(serialize=False, max_length=32, primary_key=True, editable=False),
),
]
+20 -12
View File
@@ -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()
@@ -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=32, 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 '<a href="{}">[{}]</a>'.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'
+1
View File
@@ -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))
+15 -17
View File
@@ -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)
return pickle.loads(data)
+17 -11
View File
@@ -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 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'),
+1 -1
View File
@@ -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()
+9 -5
View File
@@ -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,6 +87,10 @@ Reference
Database model describing an executed task
.. py:attribute:: id
An :func:`uuid.uuid4()` identifier
.. py:attribute:: name
The name of the task