new command rqclean [fix rq#218]

django_rq/management/commands/rqclean.py

@@ -0,0 +1,25 @@
+from django.core.management.base import BaseCommand
+
+from django_rq import get_queue
+
+
+class Command(BaseCommand):
+    """
+    Removes all queue jobs
+    """
+    help = __doc__
+
+    def add_arguments(self, parser):
+        parser.add_argument('--queue', '-q', dest='queue', default='default',
+                            help='Specify the queue [default]')
+
+    def handle(self, *args, **options):
+        """
+        Queues the function given with the first argument with the
+        parameters given with the rest of the argument list.
+        """
+        verbosity = int(options.get('verbosity', 1))
+        queue = get_queue(options.get('queue'))
+        queue.empty()
+        if verbosity:
+            print('Queue "%s" cleaned' % queue.name)

django_rq/tests/tests.py

@@ -305,6 +305,18 @@ def test_default_timeout(self):
         queue = get_queue('test1')
         self.assertEqual(queue._default_timeout, 400)
 
+    def test_rqclean_default(self):
+        queue = get_queue()
+        queue.enqueue(divide, 42, 1)
+        call_command("rqclean", "--verbosity", "0")
+        self.assertFalse(queue.jobs)
+
+    def test_rqclean_test(self):
+        queue = get_queue("test3")
+        queue.enqueue(divide, 42, 1)
+        call_command("rqclean", "--queue", "test3", "--verbosity", "0")
+        self.assertFalse(queue.jobs)
+
 
 @override_settings(RQ={'AUTOCOMMIT': True})
 class DecoratorTest(TestCase):