Merge pull request #79 from thread/queue-clear-cmd

Add `queue_clear` command

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)
 

django_lightweight_queue/management/commands/queue_clear.py

@@ -0,0 +1,51 @@
+import argparse
+
+from django.core.management.base import BaseCommand, CommandError
+
+from ...types import QueueName
+from ...utils import get_backend
+from ...backends.base import BackendWithClear
+
+
+class Command(BaseCommand):
+    help = """
+    Command to clear work on a redis-backed queue.
+
+    All pending jobs will be deleted from the queue. In flight jobs won't be
+    affected.
+    """  # noqa:A003 # inherited name
+
+    def add_arguments(self, parser: argparse.ArgumentParser) -> None:
+        parser.add_argument(
+            'queue',
+            action='store',
+            help="The queue to pause.",
+        )
+
+        parser.add_argument(
+            '--yes',
+            'skip_prompt',
+            action='store_true',
+            help="Skip confirmation prompt.",
+        )
+
+    def handle(self, queue: QueueName, skip_prompt: bool = False, **options: object) -> None:
+
+        backend = get_backend(queue)
+
+        if not isinstance(backend, BackendWithClear):
+            raise CommandError(
+                "Configured backend '{}.{}' doesn't support clearing".format(
+                    type(backend).__module__,
+                    type(backend).__name__,
+                ),
+            )
+
+        if not skip_prompt:
+            prompt = "Clear all jobs from queue {}) [y/N] ".format(queue)
+            choice = input(prompt).lower()
+
+            if choice != "y":
+                raise CommandError("Aborting")
+
+        backend.clear(queue)

tests/test_reliable_redis_backend.py

@@ -18,7 +18,7 @@
 from .mixins import RedisCleanupMixin
 
 
-class ReliableRedisDeduplicationTests(RedisCleanupMixin, unittest.TestCase):
+class ReliableRedisTests(RedisCleanupMixin, unittest.TestCase):
     longMessage = True
     prefix = settings.LIGHTWEIGHT_QUEUE_REDIS_PREFIX
 
@@ -60,19 +60,19 @@ def setUp(self) -> None:
             self.backend = ReliableRedisBackend()
         self.client = self.backend.client
 
-        super(ReliableRedisDeduplicationTests, self).setUp()
+        super(ReliableRedisTests, self).setUp()
 
         self.start_time = datetime.datetime.utcnow()
 
-    def test_empty_queue(self):
+    def test_deduplicate_empty_queue(self):
         result = self.backend.deduplicate('empty-queue')
         self.assertEqual(
             (0, 0),
             result,
             "Should do nothing when queue empty",
         )
 
-    def test_single_entry_in_queue(self):
+    def test_deduplicate_single_entry_in_queue(self):
         QUEUE = 'single-job-queue'
 
         self.enqueue_job(QUEUE)
@@ -96,7 +96,7 @@ def test_single_entry_in_queue(self):
             "Should still be a single entry in the queue",
         )
 
-    def test_unique_entries_in_queue(self):
+    def test_deduplicate_unique_entries_in_queue(self):
         QUEUE = 'unique-jobs-queue'
 
         self.enqueue_job(QUEUE, args=('args1',))
@@ -121,7 +121,7 @@ def test_unique_entries_in_queue(self):
             "Should still be a single entry in the queue",
         )
 
-    def test_duplicate_entries_in_queue(self):
+    def test_deduplicate_duplicate_entries_in_queue(self):
         QUEUE = 'duplicate-jobs-queue'
 
         self.enqueue_job(QUEUE)
@@ -146,7 +146,7 @@ def test_duplicate_entries_in_queue(self):
             "Should still be a single entry in the queue",
         )
 
-    def test_preserves_order_with_fixed_timestamps(self):
+    def test_deduplicate_preserves_order_with_fixed_timestamps(self):
         QUEUE = 'job-queue'
         WORKER_NUMBER = 0
 
@@ -201,7 +201,7 @@ def test_preserves_order_with_fixed_timestamps(self):
             "Third job dequeued should be the third job enqueued",
         )
 
-    def test_preserves_order_with_unique_timestamps(self):
+    def test_deduplicate_preserves_order_with_unique_timestamps(self):
         QUEUE = 'job-queue'
         WORKER_NUMBER = 0
 
@@ -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)