rqflush [see rq#218]

README.rst

@@ -155,6 +155,18 @@ with the path to your queue class::
 
     python manage.py rqworker high default low --queue-class 'path.to.CustomQueue'
 
+Queues cleanup
+---------------
+django_rq provides a management command that flushes the queue specified as argument::
+
+    python manage.py rqflush --queue high
+
+If you don't specify any queue it will flush the default
+
+You can suppress confirmation message if you use the option --noinput
+
+    python manage.py rqflush --queue high --noinput
+
 Support for RQ Scheduler
 ------------------------
 

django_rq/management/commands/rqflush.py

@@ -0,0 +1,42 @@
+from django.core.management.base import BaseCommand
+
+from django.utils.six.moves import input
+
+from django_rq import get_queue
+
+
+class Command(BaseCommand):
+    """
+    Flushes the queue specified as argument
+    """
+    help = __doc__
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            '--noinput', '--no-input', action='store_false',
+            dest='interactive', default=True,
+            help='Tells Django to NOT prompt the user for input of any kind.',
+        )
+        parser.add_argument('--queue', '-q', dest='queue', default='default',
+                            help='Specify the queue [default]')
+
+    def handle(self, *args, **options):
+        verbosity = int(options.get('verbosity', 1))
+        interactive = options['interactive']
+        queue = get_queue(options.get('queue'))
+
+        if interactive:
+            confirm = input("""You have requested a flush the "%s" queue.
+Are you sure you want to do this?
+
+    Type 'yes' to continue, or 'no' to cancel: """ % queue.name)
+        else:
+            confirm = 'yes'
+
+        if confirm == 'yes':
+            queue.empty()
+            if verbosity:
+                print('Queue "%s" flushed.' % queue.name)
+        else:
+            if verbosity:
+                print("Flush cancelled.")

django_rq/tests/tests.py

@@ -1,3 +1,5 @@
+import sys
+
 from django.contrib.auth.models import User
 from django.core.management import call_command
 from django.core.urlresolvers import reverse
@@ -12,6 +14,7 @@
     from django.test import override_settings
 except ImportError:
     from django.test.utils import override_settings
+from django.utils.six import StringIO
 from django.conf import settings
 
 from rq import get_current_job, Queue
@@ -79,6 +82,16 @@ def get_queue_index(name='default'):
     return queue_index
 
 
+def stub_stdin(testcase_inst, inputs):
+    stdin = sys.stdin
+
+    def cleanup():
+        sys.stdin = stdin
+
+    testcase_inst.addCleanup(cleanup)
+    sys.stdin = StringIO(inputs)
+
+
 @override_settings(RQ={'AUTOCOMMIT': True})
 class QueuesTest(TestCase):
 
@@ -306,18 +319,32 @@ def test_default_timeout(self):
         queue = get_queue('test1')
         self.assertEqual(queue._default_timeout, 400)
 
-    def test_rqclean_default(self):
+    def test_rqflush_default(self):
         queue = get_queue()
         queue.enqueue(divide, 42, 1)
-        call_command("rqclean", "--verbosity", "0")
+        call_command("rqflush", "--verbosity", "0", "--noinput")
+        self.assertFalse(queue.jobs)
+
+    def test_rqflush_test3(self):
+        queue = get_queue("test3")
+        queue.enqueue(divide, 42, 1)
+        call_command("rqflush", "--queue", "test3", "--verbosity", "0", "--noinput")
         self.assertFalse(queue.jobs)
 
-    def test_rqclean_test(self):
+    def test_rqflush_test3_interactive_yes(self):
         queue = get_queue("test3")
         queue.enqueue(divide, 42, 1)
-        call_command("rqclean", "--queue", "test3", "--verbosity", "0")
+        stub_stdin(self, "yes")
+        call_command("rqflush", "--queue", "test3", "--verbosity", "0")
         self.assertFalse(queue.jobs)
 
+    def test_rqflush_test3_interactive_no(self):
+        queue = get_queue("test3")
+        queue.enqueue(divide, 42, 1)
+        stub_stdin(self, "no")
+        call_command("rqflush", "--queue", "test3", "--verbosity", "0")
+        self.assertTrue(queue.jobs)
+
 
 @override_settings(RQ={'AUTOCOMMIT': True})
 class DecoratorTest(TestCase):