fix:defaulting to JobExecution

Author: cunla

scheduler/management/commands/rqworker.py

@@ -5,6 +5,7 @@
 import click
 from django.core.management.base import BaseCommand
 from django.db import connections
+from django.template.defaultfilters import default
 from redis.exceptions import ConnectionError
 from rq.logutils import setup_loghandlers
 

scheduler/rq_classes.py

@@ -71,16 +71,11 @@ def stop_execution(self, connection: Redis):
 class DjangoWorker(Worker):
     def __init__(self, *args, **kwargs):
         self.fork_job_execution = kwargs.pop("fork_job_execution", True)
-        job_class = kwargs.get("job_class", JobExecution)
+        job_class = kwargs.get("job_class") or JobExecution
         if not isinstance(job_class, type) or not issubclass(job_class, JobExecution):
-            if isinstance(job_class, type):
-                job_class = type("JobExecutionWrapper", (JobExecution, job_class), {})
-            else:
-                job_class = JobExecution
+            raise ValueError("job_class must be a subclass of JobExecution")
 
         # Update kwargs with the potentially modified job_class
-        kwargs["job_class"] = job_class
-
         kwargs["job_class"] = job_class
         kwargs["queue_class"] = DjangoQueue
         super(DjangoWorker, self).__init__(*args, **kwargs)

scheduler/tests/test_mgmt_cmds.py

@@ -36,6 +36,39 @@ def test_rqworker__no_queues_params(self):
         for job in jobs:
             self.assertTrue(job.is_failed)
 
+    def test_rqworker__job_class_param__green(self):
+        queue = get_queue('default')
+
+        # enqueue some jobs that will fail
+        jobs = []
+        job_ids = []
+        for _ in range(0, 3):
+            job = queue.enqueue(failing_job)
+            jobs.append(job)
+            job_ids.append(job.id)
+
+        # Create a worker to execute these jobs
+        call_command('rqworker', '--job-class', 'scheduler.rq_classes.JobExecution', fork_job_execution=False, burst=True)
+
+        # check if all jobs are really failed
+        for job in jobs:
+            self.assertTrue(job.is_failed)
+
+    def test_rqworker__bad_job_class__fail(self):
+        queue = get_queue('default')
+
+        # enqueue some jobs that will fail
+        jobs = []
+        job_ids = []
+        for _ in range(0, 3):
+            job = queue.enqueue(failing_job)
+            jobs.append(job)
+            job_ids.append(job.id)
+
+        # Create a worker to execute these jobs
+        with self.assertRaises(ImportError):
+            call_command('rqworker', '--job-class', 'rq.badclass', fork_job_execution=False, burst=True)
+
     def test_rqworker__run_jobs(self):
         queue = get_queue('default')
 

scheduler/tests/test_worker.py

@@ -43,7 +43,7 @@ def test_create_worker__scheduler_interval(self):
 
     def test_get_worker_with_custom_job_class(self):
         # Test with string representation of job_class
-        worker = create_worker('default', job_class='rq.job.Job')
+        worker = create_worker('default', job_class='scheduler.rq_classes.JobExecution')
         self.assertTrue(issubclass(worker.job_class, Job))
         self.assertTrue(issubclass(worker.job_class, JobExecution))
 

scheduler/tools.py

@@ -74,8 +74,12 @@ def create_worker(*queue_names, **kwargs):
     kwargs['name'] = kwargs['name'].replace('/', '.')
 
     # Handle job_class if provided
-    if 'job_class' in kwargs and isinstance(kwargs['job_class'], str):
+    if 'job_class' not in kwargs or kwargs["job_class"] is None:
+        kwargs['job_class'] = 'scheduler.rq_classes.JobExecution'
+    try:
         kwargs['job_class'] = import_string(kwargs['job_class'])
+    except ImportError:
+        raise ImportError(f"Could not import job class {kwargs['job_class']}")
 
     worker = DjangoWorker(queues, connection=queues[0].connection, **kwargs)
     return worker