Queue names

Author: AngryLawyer

django_dbq/management/commands/create_job.py

@@ -44,4 +44,4 @@ def handle(self, *args, **options):
             kwargs['queue_name'] = queue_name
 
         job = Job.objects.create(**kwargs)
-        self.stdout.write('Created job: "%s", id=%s' % (job.name, job.pk))
+        self.stdout.write('Created job: "%s", id=%s for queue "%s"' % (job.name, job.pk, queue_name if queue_name else 'default'))

django_dbq/management/commands/worker.py

@@ -10,15 +10,15 @@
 logger = logging.getLogger(__name__)
 
 
-def process_job():
-    """This function grabs the next available job, and runs its next task."""
+def process_job(queue_name):
+    """This function grabs the next available job for a given queue, and runs its next task."""
 
     with transaction.atomic():
-        job = Job.objects.get_ready_or_none()
+        job = Job.objects.get_ready_or_none(queue_name)
         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)
+        logger.info('Processing job: name="%s" queue="%s" id=%s state=%s next_task=%s', job.name, queue_name, job.pk, job.state, job.next_task)
         job.state = Job.STATES.PROCESSING
         job.save()
 
@@ -55,16 +55,26 @@ class Worker(WorkerProcessBase):
 
     process_title = "jobworker"
 
+    def __init__(self, name):
+        self.queue_name = name
+        super(Worker, self).__init__()
+
     def do_work(self):
         sleep(1)
-        process_job()
+        process_job(self.queue_name)
 
 
 class Command(NoArgsCommand):
 
     help = "Run a queue worker process"
 
     def handle_noargs(self, **options):
+        self.handle('default')
+
+    def handle(self, *args, **options):
+        if len(args) != 1:
+            raise CommandError("Please supply a single queue job name")
+
         logger.info("Starting job worker")
         worker = Worker()
         worker.run()

django_dbq/models.py

@@ -12,9 +12,9 @@
 
 class JobManager(models.Manager):
 
-    def get_ready_or_none(self, max_retries=3):
+    def get_ready_or_none(self, queue_name, max_retries=3):
         """
-        Get a job in state READY or NEW. Supports retrying in case of database deadlock
+        Get a job in state READY or NEW for a given queue. Supports retrying in case of database deadlock
 
         See https://dev.mysql.com/doc/refman/5.0/en/innodb-deadlocks.html
 
@@ -31,7 +31,7 @@ def get_ready_or_none(self, max_retries=3):
         retries_left = max_retries
         while True:
             try:
-                return self.select_for_update().filter(state__in=(Job.STATES.READY, Job.STATES.NEW)).first()
+                return self.select_for_update().filter(queue_name=queue_name, state__in=(Job.STATES.READY, Job.STATES.NEW)).first()
             except Exception as e:
                 if retries_left == 0:
                     raise

django_dbq/tests.py

@@ -66,25 +66,25 @@ def test_create_job(self):
         self.assertEqual(job.state, Job.STATES.NEW)
 
     def test_get_next_ready_job(self):
-        self.assertTrue(Job.objects.get_ready_or_none() is None)
+        self.assertTrue(Job.objects.get_ready_or_none('default') is None)
 
         Job.objects.create(name='testjob', state=Job.STATES.READY, created=datetime.now())
         Job.objects.create(name='testjob', state=Job.STATES.PROCESSING, created=datetime.now())
         expected = Job.objects.create(name='testjob', state=Job.STATES.READY, created=datetime.now() - timedelta(minutes=1))
 
-        self.assertEqual(Job.objects.get_ready_or_none(), expected)
+        self.assertEqual(Job.objects.get_ready_or_none('default'), expected)
 
     def test_get_next_ready_job_created(self):
         """
         Created jobs should be picked too
         """
-        self.assertTrue(Job.objects.get_ready_or_none() is None)
+        self.assertTrue(Job.objects.get_ready_or_none('default') is None)
 
         Job.objects.create(name='testjob', state=Job.STATES.NEW, created=datetime.now())
         Job.objects.create(name='testjob', state=Job.STATES.PROCESSING, created=datetime.now())
         expected = Job.objects.create(name='testjob', state=Job.STATES.NEW, created=datetime.now() - timedelta(minutes=1))
 
-        self.assertEqual(Job.objects.get_ready_or_none(), expected)
+        self.assertEqual(Job.objects.get_ready_or_none('default'), expected)
 
 
 @override_settings(JOBS={'testjob': {'tasks': ['a', 'b', 'c']}})
@@ -106,7 +106,7 @@ class ProcessJobTestCase(TestCase):
 
     def test_process_job(self):
         job = Job.objects.create(name='testjob')
-        process_job()
+        process_job('default')
         job = Job.objects.get()
         self.assertEqual(job.state, Job.STATES.COMPLETE)
 
@@ -133,7 +133,7 @@ class JobFailureHookTestCase(TestCase):
 
     def test_failure_hook(self):
         job = Job.objects.create(name='testjob')
-        process_job()
+        process_job('default')
         job = Job.objects.get()
         self.assertEqual(job.state, Job.STATES.FAILED)
         self.assertEqual(job.workspace['output'], 'failure hook ran')