Remove RQ dependency

Author: cunla

pyproject.toml

@@ -77,12 +77,3 @@ quote-style = "double"
 indent-style = "space"
 skip-magic-trailing-comma = false
 line-ending = "auto"
-#
-#[tool.coverage]
-#
-#[tool.coverage.run]
-#parallel = true
-#omit = [
-#    "**/scheduler/migrations/*"
-#]
-#concurrency = ["multiprocessing", "thread"]

scheduler/admin/task_admin.py

@@ -67,7 +67,7 @@ class Media:
     list_display = (
         "enabled",
         "name",
-        "job_id",
+        "job_name",
         "function_string",
         "is_scheduled",
         "queue",
@@ -80,7 +80,7 @@ class Media:
     )
     list_display_links = ("name",)
     readonly_fields = (
-        "job_id",
+        "job_name",
         "successful_runs",
         "last_successful_run",
         "failed_runs",
@@ -99,7 +99,7 @@ class Media:
         (None, dict(fields=("scheduled_time",), classes=("tasktype-OnceTaskType",)),),
         (None, dict(fields=("cron_string",), classes=("tasktype-CronTaskType",)),),
         (None, dict(fields=(("interval", "interval_unit",), "repeat"), classes=("tasktype-RepeatableTaskType",)),),
-        (_("Queue settings"), dict(fields=(("queue", "at_front"), "job_id"))),
+        (_("Queue settings"), dict(fields=(("queue", "at_front"), "job_name"))),
         (_("Previous runs info"),
          dict(fields=(("successful_runs", "last_successful_run"), ("failed_runs", "last_failed_run"))),),
     )

scheduler/decorators.py

@@ -1,5 +1,5 @@
 from functools import wraps
-from typing import Any, Callable, Dict, List, Optional, Union
+from typing import Any, Callable, Dict, Optional, Union
 
 from scheduler.broker_types import ConnectionType
 from scheduler.helpers.callback import Callback

scheduler/migrations/0021_remove_task_job_id_task_job_name.py

@@ -0,0 +1,22 @@
+# Generated by Django 5.1.7 on 2025-03-24 14:30
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('scheduler', '0020_remove_repeatabletask_new_task_id_and_more'),
+    ]
+
+    operations = [
+        migrations.RemoveField(
+            model_name='task',
+            name='job_id',
+        ),
+        migrations.AddField(
+            model_name='task',
+            name='job_name',
+            field=models.CharField(blank=True, editable=False, help_text='Current job_name on queue', max_length=128, null=True, verbose_name='job name'),
+        ),
+    ]

scheduler/models/task.py

@@ -44,7 +44,7 @@ def failure_callback(job: JobModel, connection, result, *args, **kwargs):
         f"Task {task.id}/{task.name} has failed",
         "See django-admin for logs",
     )
-    task.job_id = None
+    task.job_name = None
     task.failed_runs += 1
     task.last_failed_run = timezone.now()
     task.save(schedule_job=True)
@@ -55,7 +55,7 @@ def success_callback(job: JobModel, connection: ConnectionType, result: Any, *ar
     if task is None:
         logger.warn(f"Could not find task for job {job.name}")
         return
-    task.job_id = None
+    task.job_name = None
     task.successful_runs += 1
     task.last_successful_run = timezone.now()
     task.save(schedule_job=True)
@@ -95,8 +95,8 @@ class TimeUnits(models.TextChoices):
         ),
     )
     queue = models.CharField(_("queue"), max_length=255, choices=get_queue_choices, help_text=_("Queue name"))
-    job_id = models.CharField(
-        _("job name"), max_length=128, editable=False, blank=True, null=True, help_text=_("Current job_id on queue")
+    job_name = models.CharField(
+        _("job name"), max_length=128, editable=False, blank=True, null=True, help_text=_("Current job_name on queue")
     )
     at_front = models.BooleanField(
         _("At front"),
@@ -185,18 +185,18 @@ def callable_func(self):
     @admin.display(boolean=True, description=_("is scheduled?"))
     def is_scheduled(self) -> bool:
         """Check whether a next job for this task is queued/scheduled to be executed"""
-        if self.job_id is None:  # no job_id => is not scheduled
+        if self.job_name is None:  # no job_id => is not scheduled
             return False
         # check whether job_id is in scheduled/queued/active jobs
         res = (
-                (self.job_id in self.rqueue.scheduled_job_registry.all())
-                or (self.job_id in self.rqueue.queued_job_registry.all())
-                or (self.job_id in self.rqueue.active_job_registry.all())
+                (self.job_name in self.rqueue.scheduled_job_registry.all())
+                or (self.job_name in self.rqueue.queued_job_registry.all())
+                or (self.job_name in self.rqueue.active_job_registry.all())
         )
         # If the job_id is not scheduled/queued/started,
         # update the job_id to None. (The job_id belongs to a previous run which is completed)
         if not res:
-            self.job_id = None
+            self.job_name = None
             super(Task, self).save()
         return res
 
@@ -269,9 +269,9 @@ def unschedule(self) -> bool:
 
         If a job is queued to be executed or scheduled to be executed, it will remove it.
         """
-        if self.job_id is not None:
-            self.rqueue.delete_job(self.job_id)
-            self.job_id = None
+        if self.job_name is not None:
+            self.rqueue.delete_job(self.job_name)
+            self.job_name = None
         self.save(schedule_job=False)
         return True
 
@@ -361,7 +361,7 @@ def _schedule(self) -> bool:
             when=schedule_time,
             **kwargs,
         )
-        self.job_id = job.name
+        self.job_name = job.name
         super(Task, self).save()
         return True
 

scheduler/tests/test_task_types/test_cron_task.py

@@ -34,7 +34,7 @@ def test_clean_cron_string_invalid(self):
     def test_check_rescheduled_after_execution(self):
         task = task_factory(self.task_type)
         queue = task.rqueue
-        first_run_id = task.job_id
+        first_run_id = task.job_name
         entry = JobModel.get(first_run_id, connection=queue.connection)
         self.assertIsNotNone(entry)
         queue.run_sync(entry)
@@ -44,12 +44,12 @@ def test_check_rescheduled_after_execution(self):
         self.assertEqual(task.successful_runs, 1)
         self.assertIsNotNone(task.last_successful_run)
         self.assertTrue(task.is_scheduled())
-        self.assertNotEqual(task.job_id, first_run_id)
+        self.assertNotEqual(task.job_name, first_run_id)
 
     def test_check_rescheduled_after_failed_execution(self):
         task = task_factory(self.task_type, callable_name="scheduler.tests.jobs.failing_job")
         queue = task.rqueue
-        first_run_id = task.job_id
+        first_run_id = task.job_name
         entry = JobModel.get(first_run_id, connection=queue.connection)
         queue.run_sync(entry)
         task.refresh_from_db()
@@ -58,7 +58,7 @@ def test_check_rescheduled_after_failed_execution(self):
         self.assertEqual(task.successful_runs, 0)
         self.assertIsNone(task.last_successful_run)
         self.assertTrue(task.is_scheduled())
-        self.assertNotEqual(task.job_id, first_run_id)
+        self.assertNotEqual(task.job_name, first_run_id)
 
     def test_cron_task_enqueuing_jobs(self):
         queue = get_queue()
@@ -67,7 +67,7 @@ def test_cron_task_enqueuing_jobs(self):
 
         task = task_factory(self.task_type, callable_name="scheduler.tests.jobs.enqueue_jobs")
         self.assertEqual(prev_queued + 1, queue.scheduled_job_registry.count(connection=queue.connection))
-        first_run_id = task.job_id
+        first_run_id = task.job_name
         entry = JobModel.get(first_run_id, connection=queue.connection)
         queue.run_sync(entry)
         self.assertEqual(20, len(queue.queued_job_registry))

scheduler/tests/test_task_types/test_repeatable_task.py

@@ -150,7 +150,7 @@ def test_repeat_none_interval_2_min(self):
     def test_check_rescheduled_after_execution(self):
         task = task_factory(self.task_type, scheduled_time=timezone.now() + timedelta(seconds=1), repeat=10)
         queue = task.rqueue
-        first_run_id = task.job_id
+        first_run_id = task.job_name
         entry = JobModel.get(first_run_id, connection=queue.connection)
         queue.run_sync(entry)
         task.refresh_from_db()
@@ -159,7 +159,7 @@ def test_check_rescheduled_after_execution(self):
         self.assertEqual(task.successful_runs, 1)
         self.assertIsNotNone(task.last_successful_run)
         self.assertTrue(task.is_scheduled())
-        self.assertNotEqual(task.job_id, first_run_id)
+        self.assertNotEqual(task.job_name, first_run_id)
 
     def test_check_rescheduled_after_execution_failed_job(self):
         task = task_factory(
@@ -169,7 +169,7 @@ def test_check_rescheduled_after_execution_failed_job(self):
             repeat=10,
         )
         queue = task.rqueue
-        first_run_id = task.job_id
+        first_run_id = task.job_name
         entry = JobModel.get(first_run_id, connection=queue.connection)
         queue.run_sync(entry)
         task.refresh_from_db()
@@ -178,7 +178,7 @@ def test_check_rescheduled_after_execution_failed_job(self):
         self.assertEqual(task.successful_runs, 0)
         self.assertIsNone(task.last_successful_run)
         self.assertTrue(task.is_scheduled())
-        self.assertNotEqual(task.job_id, first_run_id)
+        self.assertNotEqual(task.job_name, first_run_id)
 
     def test_check_not_rescheduled_after_last_repeat(self):
         task = task_factory(
@@ -187,12 +187,12 @@ def test_check_not_rescheduled_after_last_repeat(self):
             repeat=1,
         )
         queue = task.rqueue
-        first_run_id = task.job_id
+        first_run_id = task.job_name
         entry = JobModel.get(first_run_id, connection=queue.connection)
         queue.run_sync(entry)
         task.refresh_from_db()
         self.assertEqual(task.failed_runs, 0)
         self.assertIsNone(task.last_failed_run)
         self.assertEqual(task.successful_runs, 1)
         self.assertIsNotNone(task.last_successful_run)
-        self.assertNotEqual(task.job_id, first_run_id)
+        self.assertNotEqual(task.job_name, first_run_id)

scheduler/tests/test_task_types/test_task_model.py

@@ -115,35 +115,35 @@ def test_schedule(self):
                 self.task_type,
             )
             self.assertTrue(task.is_scheduled())
-            self.assertIsNotNone(task.job_id)
+            self.assertIsNotNone(task.job_name)
 
         def test_unschedulable(self):
             task = task_factory(self.task_type, enabled=False)
             self.assertFalse(task.is_scheduled())
-            self.assertIsNone(task.job_id)
+            self.assertIsNone(task.job_name)
 
         def test_unschedule(self):
             task = task_factory(self.task_type)
             self.assertTrue(task.unschedule())
-            self.assertIsNone(task.job_id)
+            self.assertIsNone(task.job_name)
 
         def test_unschedule_not_scheduled(self):
             task = task_factory(self.task_type, enabled=False)
             self.assertTrue(task.unschedule())
-            self.assertIsNone(task.job_id)
+            self.assertIsNone(task.job_name)
 
         def test_save_enabled(self):
             task = task_factory(self.task_type)
-            self.assertIsNotNone(task.job_id)
+            self.assertIsNotNone(task.job_name)
 
         def test_save_disabled(self):
             task = task_factory(self.task_type, enabled=False)
             task.save()
-            self.assertIsNone(task.job_id)
+            self.assertIsNone(task.job_name)
 
         def test_save_and_schedule(self):
             task = task_factory(self.task_type)
-            self.assertIsNotNone(task.job_id)
+            self.assertIsNotNone(task.job_name)
             self.assertTrue(task.is_scheduled())
 
         def test_schedule2(self):
@@ -155,7 +155,7 @@ def test_schedule2(self):
 
         def test_delete_and_unschedule(self):
             task = task_factory(self.task_type)
-            self.assertIsNotNone(task.job_id)
+            self.assertIsNotNone(task.job_name)
             self.assertTrue(task.is_scheduled())
             task.delete()
             self.assertFalse(task.is_scheduled())
@@ -187,7 +187,7 @@ def test_at_front_passthrough(self):
             task = task_factory(self.task_type, at_front=True)
             queue = task.rqueue
             jobs_to_schedule = queue.scheduled_job_registry.all()
-            self.assertIn(task.job_id, jobs_to_schedule)
+            self.assertIn(task.job_name, jobs_to_schedule)
 
         def test_callable_result(self):
             task = task_factory(self.task_type)
@@ -320,7 +320,7 @@ def test_admin_run_job_now_enqueues_job_at(self):
             task.refresh_from_db()
             queue = get_queue(task.queue)
             assert_has_execution_with_status(task, JobStatus.QUEUED)
-            self.assertIn(task.job_id, queue.scheduled_job_registry.all())
+            self.assertIn(task.job_name, queue.scheduled_job_registry.all())
 
 
         def test_admin_change_view(self):
@@ -362,8 +362,8 @@ def test_admin_enqueue_job_now(self):
             # arrange
             self.client.login(username="admin", password="admin")
             task = task_factory(self.task_type)
-            self.assertIsNotNone(task.job_id)
-            job = JobModel.get(task.job_id, connection=task.rqueue.connection)
+            self.assertIsNotNone(task.job_name)
+            job = JobModel.get(task.job_name, connection=task.rqueue.connection)
             self.assertEqual(job.status, JobStatus.SCHEDULED)
             self.assertTrue(task.is_scheduled())
             data = {
@@ -399,7 +399,7 @@ def test_admin_enable_job(self):
             # arrange
             self.client.login(username="admin", password="admin")
             task = task_factory(self.task_type, enabled=False)
-            self.assertIsNone(task.job_id)
+            self.assertIsNone(task.job_name)
             self.assertFalse(task.is_scheduled())
             data = {
                 "action": "enable_selected",
@@ -446,7 +446,7 @@ def test_admin_single_delete(self):
             self.client.login(username="admin", password="admin")
             prev_count = Task.objects.filter(task_type=self.task_type).count()
             task = task_factory(self.task_type)
-            self.assertIsNotNone(task.job_id)
+            self.assertIsNotNone(task.job_name)
             self.assertTrue(task.is_scheduled())
             prev_executions_count = len(_get_executions(task))
             url = reverse("admin:scheduler_task_delete", args=[task.pk])
@@ -465,7 +465,7 @@ def test_admin_delete_selected(self):
             task.save()
             queue = get_queue(task.queue)
             scheduled_jobs = queue.scheduled_job_registry.all()
-            job_name = task.job_id
+            job_name = task.job_name
             self.assertIn(job_name, scheduled_jobs)
             data = {
                 "action": "delete_selected",

scheduler/tests/test_worker/test_scheduler.py

@@ -16,15 +16,15 @@ def test_create_worker_with_scheduler__scheduler_started(self):
 
     def test_scheduler_schedules_tasks(self):
         task = task_factory(TaskType.ONCE, scheduled_time=datetime.now() + timedelta(milliseconds=40))
-        self.assertIsNotNone(task.job_id)
-        self.assertNotIn(task.job_id, task.rqueue.queued_job_registry)
-        self.assertIn(task.job_id, task.rqueue.scheduled_job_registry)
+        self.assertIsNotNone(task.job_name)
+        self.assertNotIn(task.job_name, task.rqueue.queued_job_registry)
+        self.assertIn(task.job_name, task.rqueue.scheduled_job_registry)
         worker = create_worker("default", name="test", burst=True, with_scheduler=True)
         worker.bootstrap()
         time.sleep(1)
         worker.scheduler.enqueue_scheduled_jobs()
-        self.assertIsNotNone(task.job_id)
-        self.assertNotIn(task.job_id, task.rqueue.scheduled_job_registry)
-        self.assertIn(task.job_id, task.rqueue.queued_job_registry)
+        self.assertIsNotNone(task.job_name)
+        self.assertNotIn(task.job_name, task.rqueue.scheduled_job_registry)
+        self.assertIn(task.job_name, task.rqueue.queued_job_registry)
         worker.stop_scheduler()
         self.assertIsNone(worker.scheduler)

scheduler/tests/testtools.py

@@ -109,7 +109,7 @@ def taskarg_factory(cls, **kwargs):
 
 def _get_task_scheduled_job_from_registry(django_task: Task) -> JobModel:
     jobs_to_schedule = django_task.rqueue.scheduled_job_registry.all()
-    entry = next(i for i in jobs_to_schedule if i == django_task.job_id)
+    entry = next(i for i in jobs_to_schedule if i == django_task.job_name)
     return JobModel.get(entry, connection=django_task.rqueue.connection)