Base backend

Author: danpalmer

django_lightweight_queue/backends/base.py

@@ -0,0 +1,23 @@
+from abc import ABCMeta, abstractmethod
+
+from ..job import Job
+
+
+class BaseBackend(metaclass=ABCMeta):
+    def startup(self, queue: str) -> None:
+        pass
+
+    @abstractmethod
+    def enqueue(self, job: Job, queue: str) -> None:
+        raise NotImplementedError()
+
+    @abstractmethod
+    def dequeue(self, queue: str, worker_num: int, timeout: float) -> Job:
+        raise NotImplementedError()
+
+    @abstractmethod
+    def length(self, queue: str) -> int:
+        raise NotImplementedError()
+
+    def processed_job(self, queue: str, worker_num: int, job: Job) -> None:
+        pass

django_lightweight_queue/backends/debug_web.py

@@ -3,8 +3,10 @@
 from django.conf import settings
 from django.shortcuts import reverse
 
+from .base import BaseBackend
 
-class DebugWebBackend(object):
+
+class DebugWebBackend(BaseBackend):
     """
     This backend aids debugging in concert with the 'debug-run' view.
 
@@ -14,8 +16,6 @@ class DebugWebBackend(object):
 
     See the docstring of that view for information (and limitations) about it.
     """
-    def startup(self, queue):
-        pass
 
     def enqueue(self, job, queue):
         path = reverse('django-lightweight-queue:debug-run')
@@ -28,6 +28,3 @@ def dequeue(self, queue, worker_num, timeout):
 
     def length(self, queue):
         return 0
-
-    def processed_job(self, queue, worker_num, job):
-        pass

django_lightweight_queue/backends/redis.py

@@ -1,9 +1,11 @@
 import redis
 
-from ..job import Job
 from .. import app_settings
+from ..job import Job
+from .base import BaseBackend
+
 
-class RedisBackend(object):
+class RedisBackend(BaseBackend):
     """
     This backend has at-most-once semantics.
     """
@@ -14,9 +16,6 @@ def __init__(self):
             port=app_settings.REDIS_PORT,
         )
 
-    def startup(self, queue):
-        pass
-
     def enqueue(self, job, queue):
         self.client.lpush(self._key(queue), job.to_json().encode('utf-8'))
 
@@ -31,9 +30,6 @@ def dequeue(self, queue, worker_num, timeout):
     def length(self, queue):
         return self.client.llen(self._key(queue))
 
-    def processed_job(self, queue, worker_num, job):
-        pass
-
     def _key(self, queue):
         if app_settings.REDIS_PREFIX:
             return '{}:django_lightweight_queue:{}'.format(

django_lightweight_queue/backends/reliable_redis.py

@@ -1,9 +1,11 @@
 import redis
 
-from ..job import Job
 from .. import app_settings
+from ..job import Job
+from .base import BaseBackend
+
 
-class ReliableRedisBackend(object):
+class ReliableRedisBackend(BaseBackend):
     """
     This backend manages a per-queue-per-worker 'processing' queue. E.g. if we
     had a queue called 'django_lightweight_queue:things', and two workers, we

django_lightweight_queue/backends/synchronous.py

@@ -1,14 +1,15 @@
 import time
 
-class SynchronousBackend(object):
+from .base import BaseBackend
+
+
+class SynchronousBackend(BaseBackend):
     """
     This backend has at-most-once semantics.
     """
-    def startup(self, queue):
-        pass
 
     def enqueue(self, job, queue):
-        job.run()
+        job.run(queue=queue, worker_num=0)
 
     def dequeue(self, queue, worker_num, timeout):
         # Cannot dequeue from the synchronous backend but we can emulate by
@@ -19,6 +20,3 @@ def length(self, queue):
         # The length is the number of items waiting to be processed, which can
         # be defined as always 0 for the synchronous backend
         return 0
-
-    def processed_job(self, queue, worker_num, job):
-        pass

django_lightweight_queue/job.py

@@ -53,14 +53,19 @@ def from_json(cls, val):
     def created_time_str(self):
         return self.created_time.strftime(TIME_FORMAT)
 
-    def run(self):
+    def run(self, *, queue, worker_num):
+        """
+        `queue` and `worker_num` arguments are required for context only and do
+        not change the behaviour of job execution.
+        """
+
         start = time.time()
 
         middleware = get_middleware()
 
         for instance in middleware:
             if hasattr(instance, 'process_job'):
-                instance.process_job(self)
+                instance.process_job(self, queue, worker_num)
 
         try:
             task = self.get_task_instance()

django_lightweight_queue/views.py

@@ -50,7 +50,7 @@ def debug_run(request):
     try:
         logging.root.addHandler(handler)
         with transaction.atomic():
-            result = job.run()
+            result = job.run(queue='debug', worker_num=0)
             transaction.set_rollback(rollback=True)
     finally:
         logging.root.removeHandler(handler)

django_lightweight_queue/worker.py

@@ -136,7 +136,7 @@ def process(self, backend):
         self.log.debug("Running job %s", job)
         self.set_process_title("Running job %s" % job)
 
-        if job.run() and self.touch_filename:
+        if job.run(queue=self.queue, worker_num=self.worker_num) and self.touch_filename:
             with open(self.touch_filename, 'a'):
                 os.utime(self.touch_filename, None)