fork/django-q2
1
0
Fork 0
mirror of https://github.com/django-q2/django-q2.git synced 2026-09-23 06:48:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

hooks now work

Browse Source 
refactored task as dict instead of list for readability.
This commit is contained in:
Ilan Steemers
2015-06-18 15:23:21 +02:00
parent 90b5b0d6ab
commit 5adf92eb27
6 changed files with 88 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
django_q/apps.py
-45
View File
@@ -1,51 +1,6 @@
from multiprocessing import cpu_count
from django.apps import AppConfig
from django.conf import settings
class SessionAdminConfig(AppConfig):
name = 'django_q'
verbose_name = "Django Q"
VERSION = '0.1.0'
"""
Sets the logging level for the app
"""
try:
LOG_LEVEL = settings.Q_LOG_LEVEL
except AttributeError:
LOG_LEVEL = "INFO"
"""
Using Django's secret key to sign task packages
"""
try:
SECRET_KEY = settings.SECRET_KEY
except AttributeError:
SECRET_KEY = 'omgicantbelieveyoudonthaveasecretkey'
"""
SAVE_LIMIT limits the amount of successful task executions saved to the database.
Set this to 0 for no limits.
Failures are not limited.
"""
try:
SAVE_LIMIT = settings.Q_SAVE_LIMIT
except AttributeError:
SAVE_LIMIT = 100
try:
WORKERS = settings.Q_WORKERS
except AttributeError:
WORKERS = cpu_count()
"""
Turns compression on/off for task packages
"""
try:
COMPRESSED = settings.Q_COMPRESSED
except AttributeError:
COMPRESSED = False
django_q/conf.py
+43
View File
@@ -0,0 +1,43 @@
from django.conf import settings
from multiprocessing import cpu_count
VERSION = '0.1.0'
"""
Sets the logging level for the app
"""
try:
LOG_LEVEL = settings.Q_LOG_LEVEL
except AttributeError:
LOG_LEVEL = "INFO"
"""
Using Django's secret key to sign task packages
"""
try:
SECRET_KEY = settings.SECRET_KEY
except AttributeError:
SECRET_KEY = 'omgicantbelieveyoudonthaveasecretkey'
"""
SAVE_LIMIT limits the amount of successful task executions saved to the database.
Set this to 0 for no limits.
Failures are not limited.
"""
try:
SAVE_LIMIT = settings.Q_SAVE_LIMIT
except AttributeError:
SAVE_LIMIT = 100
try:
WORKERS = settings.Q_WORKERS
except AttributeError:
WORKERS = cpu_count()
"""
Turns compression on/off for task packages
"""
try:
COMPRESSED = settings.Q_COMPRESSED
except AttributeError:
COMPRESSED = False
django_q/main.py
+35 -42
View File
@@ -14,7 +14,7 @@ import redis
from django.core import signing
from .apps import LOG_LEVEL, SECRET_KEY, SAVE_LIMIT, WORKERS, COMPRESSED, VERSION
from .conf import LOG_LEVEL, SECRET_KEY, SAVE_LIMIT, WORKERS, COMPRESSED, VERSION
from .humanhash import uuid
from .models import Task, Success
@@ -127,23 +127,19 @@ def pusher(task_queue, e):
task = task[1]
task_queue.put(task)
logger.debug('queueing {}'.format(task))
logger.info("{} stopped pushing".format(current_process().name))
logger.info("{} stopped pushing tasks".format(current_process().name))
def monitor(done_queue):
name = current_process().name
logger.info("{} monitoring at {}".format(name, current_process().pid))
logger.info("{} monitoring results at {}".format(name, current_process().pid))
for task in iter(done_queue.get, 'STOP'):
name = task[0]
func = task[1]
result = task[6]
success = task[7]
if success:
logger.info("Finished [{}:{}]".format(func, name))
if task['success']:
logger.info("Finished [{}:{}]".format(task['func'], task['name']))
else:
logger.error("Failed [{}:{}] - {}".format(func, name, result))
logger.error("Failed [{}:{}] - {}".format(task['func'], task['name'], task['result']))
save_task(task)
logger.info("{} stopped monitoring".format(name))
logger.info("{} stopped monitoring results".format(name))
def worker(task_queue, done_queue):
@@ -157,59 +153,56 @@ def worker(task_queue, done_queue):
logger.error(e)
continue
except BadSignature as e:
task[0] = task[0].rsplit(":", 1)[0]
task.append(timezone.now())
task.append(e)
task.append(False)
task['name'] = task['name'].rsplit(":", 1)[0]
task['stopped'] = timezone.now()
task['result'] = e
task['success'] = False
done_queue.put(task)
continue
func = task[1]
module, func = func.rsplit('.', 1)
args = task[2]
kwargs = task[3]
logger.info('{} processing [{}:{}]'.format(name, func, task[0]))
task.append(timezone.now())
module, func = task['func'].rsplit('.', 1)
logger.info('{} processing [{}:{}]'.format(name, task['func'], task['name']))
try:
m = importlib.import_module(module)
f = getattr(m, func)
result = f(*args, **kwargs)
task.append(result)
task.append(True)
task['result'] = f(*task['args'], **task['kwargs'])
task['stopped'] = timezone.now()
task['success'] = True
done_queue.put(task)
except Exception as e:
task.append(e)
task.append(False)
task['result'] = e
task['stopped'] = timezone.now()
task['success'] = False
done_queue.put(task)
logger.info('{} stopped working'.format(name))
logger.info('{} stopped doing work'.format(name))
def save_task(task):
if task[7] and 0 < SAVE_LIMIT < Success.objects.count():
if task['success'] and 0 < SAVE_LIMIT < Success.objects.count():
Success.objects.first().delete()
Task.objects.create(name=task[0],
func=task[1],
args=task[2],
kwargs=task[3],
started=task[4],
stopped=task[5],
result=task[6],
success=task[7])
Task.objects.create(name=task['name'],
func=task['func'],
hook=task['hook'],
args=task['args'],
kwargs=task['kwargs'],
started=task['started'],
stopped=task['stopped'],
result=task['result'],
success=task['success'])
def async(func, *args, **kwargs):
def async(func, *args, hook=None, **kwargs):
"""
Schedules a module function
[name, func, args, kwargs, started, finished, result, success]
Schedules a task
"""
name = uuid()[0]
pack = signing.dumps([name, func, args, kwargs, timezone.now()],
task = {'name': uuid()[0], 'func': func, 'hook': hook, 'args': args, 'kwargs': kwargs, 'started': timezone.now()}
pack = signing.dumps(task,
key=SECRET_KEY,
salt='django_q.q',
compress=COMPRESSED,
serializer=JSONPickleSerializer)
r.rpush(q_list, pack)
logger.debug('Pushed {}'.format(pack))
return name
return task['name']
class JSONPickleSerializer(object):
django_q/migrations/0001_initial.py
+2 -1
View File
@@ -14,9 +14,10 @@ class Migration(migrations.Migration):
migrations.CreateModel(
name='Task',
fields=[
('id', models.AutoField(serialize=False, auto_created=True, verbose_name='ID', primary_key=True)),
('id', models.AutoField(serialize=False, verbose_name='ID', primary_key=True, auto_created=True)),
('name', models.CharField(max_length=100)),
('func', models.CharField(max_length=256)),
('hook', models.CharField(max_length=256, null=True)),
('args', picklefield.fields.PickledObjectField(editable=False)),
('kwargs', picklefield.fields.PickledObjectField(editable=False)),
('result', picklefield.fields.PickledObjectField(editable=False)),
django_q/models.py
+4 -3
View File
@@ -10,6 +10,7 @@ from picklefield import PickledObjectField
class Task(models.Model):
name = models.CharField(max_length=100)
func = models.CharField(max_length=256)
hook = models.CharField(max_length=256, null=True)
args = PickledObjectField()
kwargs = PickledObjectField()
result = PickledObjectField()
@@ -30,12 +31,12 @@ class Task(models.Model):
@receiver(post_save, sender=Task)
def call_hook(sender, instance, **kwargs):
if instance.kwargs.get('hook'):
module, func = instance.kwargs.get('hook').rsplit('.', 1)
if instance.hook:
module, func = instance.hook.rsplit('.', 1)
try:
m = importlib.import_module(module)
f = getattr(m, func)
f(task=instance)
f(instance)
except Exception as e:
logger = logging.getLogger('django-q')
logger.error('return hook failed on {}'.format(instance.name))
django_q/tests/tasks.py
+4
View File
@@ -7,3 +7,7 @@ def countdown(n):
n -= 1
stop_time = time.time()
return stop_time - start_time
def result(obj):
print('RESULT HOOK: {}'.format(obj.result))