Idea here:

#159 (comment)

* created `rqcronjobs` management command
* add some tests for it
* add some documentation

Author: lechup

README.rst

@@ -167,6 +167,33 @@ You can use also use the management command ``rqscheduler`` to start the schedul
 
     python manage.py rqscheduler
 
+Support for RQ Scheduler's cron
+-------------------------------
+
+If you have `RQ Scheduler <https://github.com/ui/rq-scheduler>`_ installed,
+you can easily manage Your cron entries using ``settings.RQ_CRONJOBS``:
+
+.. code-block:: python
+
+    RQ_CRONJOBS = [
+        ('*/10 * * * *', 'whatever.function'),
+        {
+            'cron_string': '*/10 * * * *',
+            'func': 'whatever.function',
+            'timeout': 50,
+            'args': ('foo', 'bar'),
+            'kwargs': {'foo':'bar'},
+            'queue': 'default',
+        },
+    ]
+
+Use the management command ``rqcronjobs`` after each deployment to reinstall all
+cronjobs in the scheduler::
+
+    python manage.py rqcronjobs
+
+Note, that this command will remove all previous jobs in the default scheduler.
+
 Support for django-redis and django-redis-cache
 -----------------------------------------------
 

django_rq/management/commands/rqcronjobs.py

@@ -0,0 +1,75 @@
+# -*- coding: utf-8 -*-
+from optparse import make_option
+
+from django.core.management.base import BaseCommand
+
+from django_rq import get_scheduler
+from django_rq import settings
+
+
+class Command(BaseCommand):
+    help = 'Remove all cronjobs from scheduer and install new ones defined in RQ_CRONJOBS setting (by default). ' \
+           'See possible options to get more...'
+    option_list = BaseCommand.option_list + (
+        make_option(
+            '-i',
+            '--install',
+            action='store_true',
+            dest='install',
+            default=False,
+            help='Limit only to installing cronjobs defined in RQ_CRONJOBS setting.'
+        ),
+        make_option(
+            '-r',
+            '--remove',
+            action='store_true',
+            dest='remove',
+            default=False,
+            help='Limit only to removing all cronjobs from scheduler.'
+        ),
+        make_option(
+            '-l',
+            '--list',
+            action='store_true',
+            dest='list',
+            default=False,
+            help='List cronjobs defined in RQ_CRONJOBS setting and defined in scheduler.'
+        ),
+    )
+
+    def handle(self, *args, **options):
+        scheduler = get_scheduler()
+
+        if options.get('list'):
+            print('Cronjobs from scheduler:')
+            for cronjob in scheduler.get_jobs():
+                print('* {}'.format(cronjob))
+            print('')
+            print('Cronjobs defined in settings.RQ_CRONJOBS:')
+            for cronjob_entry in settings.CRONJOBS:
+                print('* {}'.format(cronjob_entry))
+            print('')
+        else:
+            reinstall = not (options.get('install') or options.get('remove'))
+
+            if reinstall or options.get('remove'):
+                print('Removed cronjobs from scheduler:')
+                for cronjob in scheduler.get_jobs():
+                    print('* {}'.format(cronjob))
+                    cronjob.delete()
+                print('')
+
+            if reinstall or options.get('install'):
+                print('Cronjobs installed from settings.RQ_CRONJOBS:')
+                for cronjob_entry in settings.CRONJOBS:
+                    if type(cronjob_entry) is dict:
+                        args = []
+                        kwargs = cronjob_entry
+                    else:
+                        args = cronjob_entry
+                        kwargs = {}
+                    cronjob = scheduler.cron(
+                        *args, **kwargs
+                    )
+                    print('* {}'.format(cronjob))
+                print('')

django_rq/settings.py

@@ -2,6 +2,7 @@
 
 from django.conf import settings
 from django.core.exceptions import ImproperlyConfigured
+from django.utils.functional import lazy
 
 from .queues import get_unique_connection_configs
 
@@ -22,3 +23,10 @@
 
 # Get exception handlers
 EXCEPTION_HANDLERS = getattr(settings, 'RQ_EXCEPTION_HANDLERS', [])
+
+
+# laizly get RQ_CRONJOBS from django settings
+# for override_settings support in tests
+def get_cronjobs():
+    return getattr(settings, 'RQ_CRONJOBS', [])
+CRONJOBS = lazy(get_cronjobs, list)()

django_rq/tests/tests.py

@@ -592,3 +592,52 @@ def test_get_queue_django_redis_cache(self):
         self.assertEqual(connection_kwargs['port'], int(cachePort))
         self.assertEqual(connection_kwargs['db'], int(cacheDBNum))
         self.assertEqual(connection_kwargs['password'], None)
+
+
+class RqcronjobsTest(TestCase):
+    RQ_CRONJOBS = [
+        ('*/10 * * * *', 'whatever.function'),
+        {
+            'cron_string': '*/10 * * * *',
+            'func': 'whatever.function',
+            'timeout': 5,
+        },
+    ]
+
+    def setUp(self):
+        self.scheduler = get_scheduler()
+        self.clear_scheduler()
+
+    def tearDown(self):
+        self.clear_scheduler()
+
+    def clear_scheduler(self):
+        for scheduled_job in self.scheduler.get_jobs():
+            scheduled_job.delete()
+
+    def add_one_job_to_scheduler(self):
+        self.scheduler.cron(
+            *self.RQ_CRONJOBS[0]
+        )
+
+    @skipIf(RQ_SCHEDULER_INSTALLED is False, 'RQ Scheduler not installed')
+    def test_remove(self):
+        self.add_one_job_to_scheduler()
+        self.assertEqual(len(self.scheduler.get_jobs()), 1)
+        call_command('rqcronjobs', remove=True)
+        self.assertEqual(len(self.scheduler.get_jobs()), 0)
+
+    @skipIf(RQ_SCHEDULER_INSTALLED is False, 'RQ Scheduler not installed')
+    @override_settings(RQ_CRONJOBS=RQ_CRONJOBS)
+    def test_install(self):
+        self.assertEqual(len(self.scheduler.get_jobs()), 0)
+        call_command('rqcronjobs', install=True)
+        self.assertEqual(len(self.scheduler.get_jobs()), 2)
+
+    @skipIf(RQ_SCHEDULER_INSTALLED is False, 'RQ Scheduler not installed')
+    @override_settings(RQ_CRONJOBS=RQ_CRONJOBS)
+    def test_reinstall(self):
+        self.add_one_job_to_scheduler()
+        self.assertEqual(len(self.scheduler.get_jobs()), 1)
+        call_command('rqcronjobs')
+        self.assertEqual(len(self.scheduler.get_jobs()), 2)