Add clear() method to supporting backends

Author: lirsacc

django_lightweight_queue/backends/base.py

@@ -54,6 +54,12 @@ def deduplicate(
         raise NotImplementedError()
 
 
+class BackendWithClear(BaseBackend, metaclass=ABCMeta):
+    @abstractmethod
+    def clear(self, queue: QueueName) -> None:
+        raise NotImplementedError()
+
+
 class BackendWithPause(BaseBackend, metaclass=ABCMeta):
     @abstractmethod
     def pause(self, queue: QueueName, until: datetime.datetime) -> None:

django_lightweight_queue/backends/redis.py

@@ -5,12 +5,12 @@
 
 from .. import app_settings
 from ..job import Job
-from .base import BackendWithPauseResume
+from .base import BackendWithClear, BackendWithPauseResume
 from ..types import QueueName, WorkerNumber
 from ..utils import block_for_time
 
 
-class RedisBackend(BackendWithPauseResume):
+class RedisBackend(BackendWithPauseResume, BackendWithClear):
     """
     This backend has at-most-once semantics.
     """
@@ -78,6 +78,9 @@ def resume(self, queue: QueueName) -> None:
     def is_paused(self, queue: QueueName) -> bool:
         return bool(self.client.exists(self._pause_key(queue)))
 
+    def clear(self, queue: QueueName) -> None:
+        self.client.delete(self._key(queue))
+
     def _key(self, queue: QueueName) -> str:
         if app_settings.REDIS_PREFIX:
             return '{}:django_lightweight_queue:{}'.format(

django_lightweight_queue/backends/reliable_redis.py

@@ -5,7 +5,11 @@
 
 from .. import app_settings
 from ..job import Job
-from .base import BackendWithDeduplicate, BackendWithPauseResume
+from .base import (
+    BackendWithClear,
+    BackendWithDeduplicate,
+    BackendWithPauseResume,
+)
 from ..types import QueueName, WorkerNumber
 from ..utils import block_for_time, get_worker_numbers
 from ..progress_logger import ProgressLogger, NULL_PROGRESS_LOGGER
@@ -15,7 +19,7 @@
 T = TypeVar('T')
 
 
-class ReliableRedisBackend(BackendWithDeduplicate, BackendWithPauseResume):
+class ReliableRedisBackend(BackendWithClear, BackendWithDeduplicate, BackendWithPauseResume):
     """
     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
@@ -228,6 +232,9 @@ def resume(self, queue: QueueName) -> None:
     def is_paused(self, queue: QueueName) -> bool:
         return bool(self.client.exists(self._pause_key(queue)))
 
+    def clear(self, queue: QueueName) -> None:
+        self.client.delete(self._key(queue))
+
     def _key(self, queue: QueueName) -> str:
         key = 'django_lightweight_queue:{}'.format(queue)
 

tests/test_reliable_redis_backend.py

@@ -348,3 +348,13 @@ def test_pause(self):
 
         self.assertIsNone(job, "Should have indicated no work was done")
         self.assertTrue(mock_sleep.called)
+
+    def test_clear(self):
+        QUEUE = 'the-queue'
+
+        self.enqueue_job(QUEUE)
+
+        self.backend.clear(QUEUE)
+
+        dequeued = self.backend.dequeue(QUEUE, worker_number=3, timeout=1)
+        self.assertIsNone(dequeued)