Support for RQ's Cron Scheduler (WIP) (#725)

selwin · web-flow · commit 6b1ac3b44080 · 2025-10-12T20:01:46.000+07:00
* Created DjangoCronScheduler

* Successfully create rqcron management command

* Cleaned up test

* Require RQ &gt;= 2.6

* Typing fixes and bump minimum Django and Python requirement

* Fix mypy

* Fix mypy
diff --git a/django_rq/cron.py b/django_rq/cron.py
@@ -0,0 +1,126 @@
+import logging
+from typing import Any, Callable, Dict, Optional, Tuple
+
+from rq.cron import CronScheduler
+
+from .queues import get_connection
+
+
+class DjangoCronScheduler(CronScheduler):
+    """
+    A Django-RQ bridge for RQ's CronScheduler that integrates with django_rq's
+    queue configuration system.
+
+    Key differences from RQ's CronScheduler:
+    - Can be initialized without a connection parameter
+    - Connection is set dynamically when the first job is registered
+    - Validates that all registered jobs use queues with the same Redis connection
+    - Integrates with RQ_QUEUES configuration from Django settings
+    """
+
+    _connection_config: Optional[Dict[str, Any]]
+
+    def __init__(self, logging_level: int = logging.INFO, name: str = ""):
+        """
+        Initialize DjangoCronScheduler without Redis connection.
+
+        Connection will be set when the first job is registered via register().
+
+        Args:
+            logging_level: Logging level for the scheduler
+            name: Optional name for the scheduler instance
+        """
+        # Call parent __init__ with connection=None initially
+        super().__init__(connection=None, logging_level=logging_level)
+
+        # Track our django_rq specific state
+        self._connection_config = None
+
+    def _get_connection_config(self, queue_name: str) -> Dict[str, Any]:
+        """
+        Extract Redis connection configuration for a queue to compare connections.
+
+        Args:
+            queue_name: Name of the queue
+
+        Returns:
+            Dictionary of connection parameters for comparison
+        """
+        connection = get_connection(queue_name)
+        kwargs = connection.connection_pool.connection_kwargs
+
+        # Only compare essential connection parameters that determine if
+        # two connections are to the same Redis instance
+        essential_params = ["host", "port", "db", "username", "password"]
+        return {key: kwargs.get(key) for key in essential_params if key in kwargs}
+
+    def register(
+        self,
+        func: Callable[..., Any],
+        queue_name: str,
+        args: Optional[Tuple[Any, ...]] = None,
+        kwargs: Optional[Dict[str, Any]] = None,
+        interval: Optional[int] = None,
+        cron: Optional[str] = None,
+        timeout: Optional[int] = None,
+        result_ttl: int = 500,
+        ttl: Optional[int] = None,
+        failure_ttl: Optional[int] = None,
+        meta: Optional[Dict[str, Any]] = None,
+    ):
+        """
+        Register a function to be run at regular intervals.
+
+        On first call, this sets the Redis connection for the scheduler.
+        Subsequent calls validate that the queue uses the same Redis connection.
+
+        Args:
+            func: Function to be scheduled
+            queue_name: Name of the django_rq queue (must exist in RQ_QUEUES)
+            args: Arguments to pass to the function
+            kwargs: Keyword arguments to pass to the function
+            interval: Interval in seconds (mutually exclusive with cron)
+            cron: Cron expression (mutually exclusive with interval)
+            timeout: Job timeout in seconds
+            result_ttl: How long to keep job results
+            ttl: Job time-to-live
+            failure_ttl: How long to keep failed job info
+            meta: Additional job metadata
+
+        Returns:
+            CronJob instance
+
+        Raises:
+            ValueError: If queue not found or uses different Redis connection
+        """
+        # Get connection for this queue
+        connection = get_connection(queue_name)
+        current_config = self._get_connection_config(queue_name)
+
+        if self._connection_config:
+            # Validate that this queue uses the same Redis connection
+            if current_config != self._connection_config:
+                raise ValueError(
+                    f"Queue '{queue_name}' uses a different Redis connection than previously "
+                    + "registered queues. All jobs in a DjangoCronScheduler instance must use "
+                    + "queues with the same Redis connection."
+                )
+        else:
+            # First registration - set connection
+            self.connection = connection
+            self._connection_config = current_config
+
+        # Now call parent register method
+        return super().register(
+            func=func,
+            queue_name=queue_name,
+            args=args,
+            kwargs=kwargs,
+            interval=interval,
+            cron=cron,
+            timeout=timeout,
+            result_ttl=result_ttl,
+            ttl=ttl,
+            failure_ttl=failure_ttl,
+            meta=meta,
+        )
diff --git a/django_rq/management/commands/rqcron.py b/django_rq/management/commands/rqcron.py
@@ -0,0 +1,59 @@
+import logging
+import sys
+from typing import Any
+
+from django.core.management.base import BaseCommand, CommandParser
+
+from ...cron import DjangoCronScheduler
+
+
+class Command(BaseCommand):
+    """
+    Starts the RQ cron scheduler with Django-RQ integration.
+
+    Example usage:
+    python manage.py rqcron cron_config.py
+    python manage.py rqcron myapp.cron_jobs --logging-level DEBUG
+    """
+
+    help = "Starts the RQ cron scheduler"
+
+    def add_arguments(self, parser: CommandParser) -> None:
+        # Positional argument for config file/module
+        parser.add_argument(
+            'config_path',
+            help='Path to cron configuration file or module path'
+        )
+        
+        # Optional logging level
+        parser.add_argument(
+            '--logging-level', '-l',
+            choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'],
+            default='INFO',
+            help='Set logging level (default: INFO)'
+        )
+
+    def handle(self, *args: Any, **options: Any) -> None:
+        """Main command handler."""
+        config_path: str = options['config_path']
+        logging_level: int = getattr(logging, options['logging_level'])
+        
+        # Create Django cron scheduler
+        scheduler = DjangoCronScheduler(logging_level=logging_level)
+        
+        try:
+            # Load configuration from file
+            self.stdout.write(f'Loading cron configuration from {config_path}')
+            scheduler.load_config_from_file(config_path)
+            
+            # Start the scheduler
+            job_count = len(scheduler.get_jobs())
+            self.stdout.write(
+                self.style.SUCCESS(f'Starting cron scheduler with {job_count} jobs...')
+            )
+            
+            scheduler.start()
+            
+        except KeyboardInterrupt:
+            self.stdout.write('\nShutting down cron scheduler...')
+            sys.exit(0)
diff --git a/django_rq/tests/cron_config1.py b/django_rq/tests/cron_config1.py
@@ -0,0 +1,13 @@
+"""
+Cron configuration file #1 for testing the rqcron management command.
+Contains 2 jobs for the main test.
+"""
+from rq import cron
+from .fixtures import say_hello
+
+
+# Register a simple cron job that runs every minute
+cron.register(say_hello, 'default', args=('from cron config1',), cron='* * * * *')
+
+# Register another job that runs every 5 seconds using interval
+cron.register(say_hello, 'default', args=('every 5 seconds config1',), interval=5)
diff --git a/django_rq/tests/cron_config2.py b/django_rq/tests/cron_config2.py
@@ -0,0 +1,13 @@
+"""
+Cron configuration file #2 for testing the rqcron management command.
+Contains different jobs for alternative test scenarios.
+"""
+from rq import cron
+from .fixtures import say_hello
+
+
+# Register a job with different arguments for testing
+cron.register(say_hello, 'default', args=('from cron config2',), cron='*/2 * * * *')  # Every 2 minutes
+
+# Register another job with a different interval
+cron.register(say_hello, 'default', args=('every 10 seconds config2',), interval=10)
diff --git a/django_rq/tests/fixtures.py b/django_rq/tests/fixtures.py
@@ -35,3 +35,8 @@ def access_self():
 
 def failing_job():
     raise ValueError
+
+
+def say_hello(name='World'):
+    """Simple test function for cron job testing."""
+    return f"Hello, {name}!"
diff --git a/django_rq/tests/test_cron.py b/django_rq/tests/test_cron.py
@@ -0,0 +1,149 @@
+import tempfile
+import os
+from contextlib import suppress
+from io import StringIO
+from unittest import TestCase
+from unittest.mock import patch
+
+from django.core.management import call_command
+from django.core.management.base import CommandError
+from rq.cron import CronJob
+
+from ..cron import DjangoCronScheduler
+from ..management.commands.rqcron import Command as RqcronCommand
+from .fixtures import say_hello
+
+
+class CronTest(TestCase):
+
+    def test_django_cron_scheduler_init(self):
+        """Test DjangoCronScheduler can be initialized without connection."""
+        scheduler = DjangoCronScheduler()
+
+        # Should not have connection until first register() call
+        self.assertIsNone(scheduler.connection)
+        self.assertIsNone(scheduler._connection_config)
+        self.assertEqual(scheduler._cron_jobs, [])
+
+    def test_first_register_initializes_connection(self):
+        """Test that first register() call initializes the scheduler with queue's connection."""
+        scheduler = DjangoCronScheduler()
+
+        # Register a job with cron expression (run every minute)
+        cron_job = scheduler.register(say_hello, 'default', cron='* * * * *')
+
+        # Should now have connection set
+        self.assertIsNotNone(scheduler.connection)
+        self.assertIsNotNone(scheduler._connection_config)
+        self.assertIsInstance(cron_job, CronJob)
+        self.assertEqual(len(scheduler.get_jobs()), 1)
+
+        # Verify cron expression is set correctly
+        self.assertEqual(cron_job.cron, '* * * * *')
+        self.assertIsNone(cron_job.interval)
+        self.assertIsNotNone(cron_job.next_run_time)
+
+    def test_connection_validation(self):
+        """Test connection validation for same, compatible, and incompatible queues."""
+        # Start with test3 queue (localhost:6379, DB=1)
+        scheduler = DjangoCronScheduler()
+
+        # Same queue multiple times should work
+        job1 = scheduler.register(say_hello, 'test3', interval=60)
+        job2 = scheduler.register(say_hello, 'test3', interval=120)
+
+        self.assertEqual(len(scheduler.get_jobs()), 2)
+        self.assertEqual(job1.queue_name, 'test3')
+        self.assertEqual(job2.queue_name, 'test3')
+
+        # Compatible queues (same Redis connection) should work
+        # Both 'test3' and 'async' use localhost:6379 with DB=1
+        job3 = scheduler.register(say_hello, 'async', interval=180)
+        self.assertEqual(len(scheduler.get_jobs()), 3)
+        self.assertEqual(job3.queue_name, 'async')
+
+        # Queues having different Redis connections should fail
+        # 'default' uses DB=0 while test3/async use DB=1
+        with self.assertRaises(ValueError):
+            scheduler.register(say_hello, 'default', interval=240)
+
+        # Undefined queue_name should fail
+        scheduler = DjangoCronScheduler()
+        with self.assertRaises(KeyError):
+            scheduler.register(say_hello, 'nonexistent_queue', interval=300)
+
+
+class CronCommandTest(TestCase):
+
+    @patch('django_rq.cron.DjangoCronScheduler.start')
+    def test_rqcron_command(self, mock_start):
+        """Test rqcron command execution: success and import errors from load_config_from_file."""
+        mock_start.return_value = None
+
+        # Test 1: Successful execution
+        out = StringIO()
+        config_path = 'django_rq.tests.cron_config1'
+
+        call_command('rqcron', config_path, stdout=out)
+
+        output = out.getvalue()
+        self.assertIn(f'Loading cron configuration from {config_path}', output)
+        self.assertIn('Starting cron scheduler with 2 jobs...', output)
+        mock_start.assert_called_once()
+
+        # Test 2: File not found - should raise ImportError from RQ
+        with self.assertRaises(ImportError) as cm:
+            call_command('rqcron', 'nonexistent_file.py')
+
+        self.assertIn("No module named 'nonexistent_file'", str(cm.exception))
+
+        # Test 3: Import error
+        with self.assertRaises(ImportError) as cm:
+            call_command('rqcron', 'nonexistent.module.path')
+
+        self.assertIn("No module named 'nonexistent'", str(cm.exception))
+
+    @patch('django_rq.cron.DjangoCronScheduler.start')
+    @patch('django_rq.cron.DjangoCronScheduler.load_config_from_file')
+    def test_rqcron_command_exceptions(self, mock_load_config, mock_start):
+        """Test rqcron command exception handling."""
+        mock_load_config.return_value = None
+
+        # Test KeyboardInterrupt handling
+        mock_start.side_effect = KeyboardInterrupt()
+        with self.assertRaises(SystemExit):
+            call_command('rqcron', 'django_rq.tests.cron_config2')
+
+        # Test general exception handling - should bubble up as raw exception
+        mock_load_config.side_effect = Exception("Test error")
+        with self.assertRaises(Exception) as cm:
+            call_command('rqcron', 'django_rq.tests.cron_config2')
+
+        self.assertEqual(str(cm.exception), "Test error")
+
+    def test_rqcron_command_successful_run(self):
+        """Test successful rqcron command execution without mocking."""
+        out = StringIO()
+        config_path = 'django_rq.tests.cron_config1'
+
+        # Use a very short timeout to test actual execution
+        import signal
+
+        def timeout_handler(signum, frame):
+            raise KeyboardInterrupt()
+
+        # Set up a timeout to stop the command after a short time
+        old_handler = signal.signal(signal.SIGALRM, timeout_handler)
+        signal.alarm(1)  # Stop after 1 second
+
+        try:
+            # The command will be interrupted and may or may not raise SystemExit depending on Django version
+            with suppress(SystemExit):
+                call_command('rqcron', config_path, stdout=out)
+        finally:
+            signal.alarm(0)  # Cancel the alarm
+            signal.signal(signal.SIGALRM, old_handler)
+
+        output = out.getvalue()
+        self.assertIn(f'Loading cron configuration from {config_path}', output)
+        self.assertIn('Starting cron scheduler with 2 jobs...', output)
diff --git a/pyproject.toml b/pyproject.toml
@@ -10,8 +10,8 @@ readme = "README.rst"
 license = "MIT"
 license-files = ["LICENSE.txt"]
 authors = [{ name = "Selwin Ong", email = "selwin.ong@gmail.com" }]
-requires-python = ">=3.8"
-dependencies = ["django>=3.2", "rq>=2", "redis>=3.5"]
+requires-python = ">=3.9"
+dependencies = ["django>=4.2", "rq>=2.6", "redis>=3.5"]
 classifiers = [
     "Development Status :: 5 - Production/Stable",
     "Environment :: Web Environment",
@@ -81,6 +81,10 @@ ignore_missing_imports = true
 module = "rq_scheduler.*"
 ignore_missing_imports = true
 
+[[tool.mypy.overrides]]
+module = "rq.cron"
+ignore_missing_imports = true
+
 [[tool.mypy.overrides]]
 module = "sentry_sdk.*"
 ignore_missing_imports = true
