Initial commit

Author: AngryLawyer

.gitignore

@@ -0,0 +1,11 @@
+*.pyc
+*.db
+.coverage
+MANIFEST
+dist/
+build/
+env/
+html/
+htmlcov/
+*.egg-info/
+.tox/

django_dbq/__init__.py

@@ -0,0 +1 @@
+__version__ = '0.0.1'

django_dbq/management/__init__.py

@@ -0,0 +1,34 @@
+from django.conf import settings
+from django.core.management.base import BaseCommand, CommandError
+from django_dbq.apps.core.models import Job
+from optparse import make_option
+import json
+import logging
+
+
+logger = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+
+    help = "Create a job"
+    args = '<job_name>'
+
+    option_list = BaseCommand.option_list + (
+        make_option('--workspace',
+                    help='JSON-formatted initial command workspace'),)
+
+    def handle(self, *args, **options):
+        if len(args) != 1:
+            raise CommandError("Please supply a single job name")
+
+        name = args[0]
+        if name not in settings.JOBS:
+            raise CommandError('"%s" is not a valid job name' % name)
+
+        workspace = options['workspace']
+        if workspace:
+            workspace = json.loads(workspace)
+
+        job = Job.objects.create(name=name, workspace=workspace)
+        self.stdout.write('Created job: "%s", id=%s' % (job.name, job.pk))

django_dbq/management/commands/__init__.py

@@ -0,0 +1,70 @@
+from django.db import transaction
+from django.core.management.base import NoArgsCommand
+from django.utils.module_loading import import_by_path
+from django_dbq.apps.core.models import Job
+from simplesignals.process import WorkerProcessBase
+from time import sleep
+import logging
+
+
+logger = logging.getLogger(__name__)
+
+
+def process_job():
+    """This function grabs the next available job, and runs its next task."""
+
+    with transaction.atomic():
+        job = Job.objects.get_ready_or_none()
+        if not job:
+            return
+
+        logger.info('Processing job: name="%s" id=%s state=%s next_task=%s', job.name, job.pk, job.state, job.next_task)
+        job.state = Job.STATES.PROCESSING
+        job.save()
+
+    try:
+        task_function = import_by_path(job.next_task)
+        task_function(job)
+        job.update_next_task()
+        if not job.next_task:
+            job.state = Job.STATES.COMPLETE
+        else:
+            job.state = Job.STATES.READY
+    except Exception as exception:
+        logger.exception("Job id=%s failed", job.pk)
+        job.state = Job.STATES.FAILED
+
+        failure_hook_name = job.get_failure_hook_name()
+        if failure_hook_name:
+            logger.info("Running failure hook %s for job id=%s", failure_hook_name, job.pk)
+            failure_hook_function = import_by_path(failure_hook_name)
+            failure_hook_function(job, exception)
+        else:
+            logger.info("No failure hook for job id=%s", job.pk)
+
+    logger.info('Updating job: name="%s" id=%s state=%s next_task=%s', job.name, job.pk, job.state, job.next_task or 'none')
+
+    try:
+        job.save()
+    except:
+        logger.error('Failed to save job: id=%s org=%s', job.pk, job.workspace.get('organisation_id'))
+        raise
+
+
+class Worker(WorkerProcessBase):
+
+    process_title = "jobworker"
+
+    def do_work(self):
+        sleep(1)
+        process_job()
+
+
+class Command(NoArgsCommand):
+
+    help = "Run a queue worker process"
+
+    def handle_noargs(self, **options):
+        logger.info("Starting job worker")
+        worker = Worker()
+        worker.run()

django_dbq/management/commands/create_job.py

@@ -0,0 +1,42 @@
+# -*- coding: utf-8 -*-
+from south.utils import datetime_utils as datetime
+from south.db import db
+from south.v2 import SchemaMigration
+from django.db import models
+
+
+class Migration(SchemaMigration):
+
+    def forwards(self, orm):
+        # Adding model 'Job'
+        db.create_table(u'core_job', (
+            ('id', self.gf('uuidfield.fields.UUIDField')(unique=True, max_length=32, primary_key=True, db_index=True)),
+            ('created', self.gf('django.db.models.fields.DateTimeField')(auto_now_add=True, db_index=True, blank=True)),
+            ('modified', self.gf('django.db.models.fields.DateTimeField')(auto_now=True, blank=True)),
+            ('name', self.gf('django.db.models.fields.CharField')(max_length=100)),
+            ('state', self.gf('django.db.models.fields.CharField')(default='READY', max_length=20, db_index=True)),
+            ('next_task', self.gf('django.db.models.fields.CharField')(max_length=100, blank=True)),
+            ('workspace', self.gf('jsonfield.fields.JSONField')(null=True)),
+        ))
+        db.send_create_signal(u'core', ['Job'])
+
+
+    def backwards(self, orm):
+        # Deleting model 'Job'
+        db.delete_table(u'core_job')
+
+
+    models = {
+        u'core.job': {
+            'Meta': {'ordering': "['-created']", 'object_name': 'Job'},
+            'created': ('django.db.models.fields.DateTimeField', [], {'auto_now_add': 'True', 'db_index': 'True', 'blank': 'True'}),
+            'id': ('uuidfield.fields.UUIDField', [], {'unique': 'True', 'max_length': '32', 'primary_key': 'True', 'db_index': 'True'}),
+            'modified': ('django.db.models.fields.DateTimeField', [], {'auto_now': 'True', 'blank': 'True'}),
+            'name': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
+            'next_task': ('django.db.models.fields.CharField', [], {'max_length': '100', 'blank': 'True'}),
+            'state': ('django.db.models.fields.CharField', [], {'default': "'READY'", 'max_length': '20', 'db_index': 'True'}),
+            'workspace': ('jsonfield.fields.JSONField', [], {'null': 'True'})
+        }
+    }
+
+    complete_apps = ['core']

django_dbq/management/commands/worker.py

@@ -0,0 +1,86 @@
+from django.db import models
+from django.utils.module_loading import import_by_path
+from django_dbq.apps.core.tasks import get_next_task_name, get_failure_hook_name, get_creation_hook_name
+from jsonfield import JSONField
+from model_utils import Choices
+import logging
+import uuidfield
+
+
+logger = logging.getLogger(__name__)
+
+
+class JobManager(models.Manager):
+
+    def get_ready_or_none(self, max_retries=3):
+        """
+        Get a job in state READY. Supports retrying in case of database deadlock
+
+        See https://dev.mysql.com/doc/refman/5.0/en/innodb-deadlocks.html
+
+        "Always be prepared to re-issue a transaction if it fails due to
+        deadlock. Deadlocks are not dangerous. Just try again."
+
+        In the `except` clause, it's difficult to be more specific on the
+        exception type, because it's different on different backends. MySQL,
+        for example, raises a django.db.utils.InternalError for all manner of
+        database-related problems. This code is more-or-less cribbed from
+        django-celery, which uses a very similar approach.
+
+        """
+        retries_left = max_retries
+        while True:
+            try:
+                return self.select_for_update().filter(state=Job.STATES.READY).first()
+            except Exception as e:
+                if retries_left == 0:
+                    raise
+                retries_left -= 1
+                logger.warn("Caught %s when looking for a READY job, retrying %s more times", str(e), retries_left)
+
+
+class Job(models.Model):
+
+    STATES = Choices("READY", "PROCESSING", "FAILED", "COMPLETE")
+
+    id = uuidfield.UUIDField(primary_key=True, auto=True, db_index=True)
+    created = models.DateTimeField(auto_now_add=True, db_index=True)
+    modified = models.DateTimeField(auto_now=True)
+    name = models.CharField(max_length=100)
+    state = models.CharField(max_length=20, choices=STATES, default=STATES.READY, db_index=True)
+    next_task = models.CharField(max_length=100, blank=True)
+    workspace = JSONField(null=True)
+
+    class Meta:
+        ordering = ['-created']
+
+    objects = JobManager()
+
+    def save(self, *args, **kwargs):
+        if not self.pk:
+            self.next_task = get_next_task_name(self.name)
+            self.workspace = self.workspace or {}
+
+            try:
+                self.run_creation_hook()
+            except Exception as exception:  # noqa
+                logger.exception("Failed to create new job, creation hook raised an exception")
+                return  # cancel the save
+
+        return super(Job, self).save(*args, **kwargs)
+
+    def update_next_task(self):
+        self.next_task = get_next_task_name(self.name, self.next_task) or ''
+
+    def get_failure_hook_name(self):
+        return get_failure_hook_name(self.name)
+
+    def get_creation_hook_name(self):
+        return get_creation_hook_name(self.name)
+
+    def run_creation_hook(self):
+        creation_hook_name = self.get_creation_hook_name()
+        if creation_hook_name:
+            logger.info("Running creation hook %s for new job", creation_hook_name)
+            creation_hook_function = import_by_path(creation_hook_name)
+            creation_hook_function(self)

django_dbq/migrations/0001_initial.py

@@ -0,0 +1,29 @@
+from django.conf import settings
+from django_dbq.apps.core.models import Job
+from rest_framework import serializers
+import json
+
+
+class JobSerializer(serializers.Serializer):
+    name = serializers.ChoiceField()
+    created = serializers.DateTimeField(read_only=True)
+    modified = serializers.DateTimeField(read_only=True)
+    state = serializers.CharField(read_only=True)
+    workspace = serializers.WritableField(required=False)
+    url = serializers.HyperlinkedIdentityField(view_name='job_detail')
+
+    def __init__(self, *args, **kwargs):
+        super(JobSerializer, self).__init__(*args, **kwargs)
+        self.fields['name'].choices = ((key, key) for key in settings.JOBS)
+
+    def validate_workspace(self, attrs, source):
+        workspace = attrs.get('workspace')
+        if workspace and isinstance(workspace, basestring):
+            try:
+                attrs['workspace'] = json.loads(workspace)
+            except ValueError:
+                raise serializers.ValidationError("Invalid JSON")
+        return attrs
+
+    def restore_object(self, attrs, instance=None):
+        return Job(**attrs)

django_dbq/migrations/__init__.py

@@ -0,0 +1,34 @@
+from django.conf import settings
+
+
+TASK_LIST_KEY = 'tasks'
+FAILURE_HOOK_KEY = 'failure_hook'
+CREATION_HOOK_KEY = 'creation_hook'
+
+
+def get_next_task_name(job_name, current_task=None):
+    """Given a job name and (optionally) a task name, return the
+    next task in the list. If the current_task is None, return the
+    first task. If current_task is the last task in the list, return None"""
+
+    task_list = settings.JOBS[job_name][TASK_LIST_KEY]
+
+    if current_task is None:
+        return task_list[0]
+
+    next_task_index = task_list.index(current_task) + 1
+
+    try:
+        return task_list[next_task_index]
+    except IndexError:
+        return None
+
+
+def get_failure_hook_name(job_name):
+    """Return the name of the failure hook for the given job (as a string) or None"""
+    return settings.JOBS[job_name].get(FAILURE_HOOK_KEY)
+
+
+def get_creation_hook_name(job_name):
+    """Return the name of the creation hook for the given job (as a string) or None"""
+    return settings.JOBS[job_name].get(CREATION_HOOK_KEY)