diff --git a/django_q/apps.py b/django_q/apps.py index e0e58a4..361609e 100644 --- a/django_q/apps.py +++ b/django_q/apps.py @@ -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 diff --git a/django_q/conf.py b/django_q/conf.py new file mode 100644 index 0000000..215f08c --- /dev/null +++ b/django_q/conf.py @@ -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 diff --git a/django_q/main.py b/django_q/main.py index 14fb4b9..7506c19 100644 --- a/django_q/main.py +++ b/django_q/main.py @@ -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): diff --git a/django_q/migrations/0001_initial.py b/django_q/migrations/0001_initial.py index b91d43e..a93a881 100644 --- a/django_q/migrations/0001_initial.py +++ b/django_q/migrations/0001_initial.py @@ -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)), diff --git a/django_q/models.py b/django_q/models.py index dcf87ed..bf36334 100644 --- a/django_q/models.py +++ b/django_q/models.py @@ -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)) diff --git a/django_q/tests/tasks.py b/django_q/tests/tasks.py index 1c6d0fe..fc1ecf8 100644 --- a/django_q/tests/tasks.py +++ b/django_q/tests/tasks.py @@ -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))