refactor

Author: cunla

scheduler/models/task.py

@@ -45,7 +45,7 @@ def failure_callback(job: JobModel, connection, result, *args, **kwargs):
     task.job_name = None
     task.failed_runs += 1
     task.last_failed_run = timezone.now()
-    task.save(schedule_job=True)
+    task.save(schedule_job=True, clean=False)
 
 
 def success_callback(job: JobModel, connection: ConnectionType, result: Any, *args, **kwargs):
@@ -56,7 +56,7 @@ def success_callback(job: JobModel, connection: ConnectionType, result: Any, *ar
     task.job_name = None
     task.successful_runs += 1
     task.last_successful_run = timezone.now()
-    task.save(schedule_job=True)
+    task.save(schedule_job=True, clean=False)
 
 
 def get_queue_choices():
@@ -266,7 +266,7 @@ def unschedule(self) -> bool:
         if self.job_name is not None:
             self.rqueue.delete_job(self.job_name)
             self.job_name = None
-        self.save(schedule_job=False)
+        self.save(schedule_job=False, clean=False)
         return True
 
     def _schedule_time(self) -> datetime:
@@ -360,7 +360,9 @@ def _schedule(self) -> bool:
         return True
 
     def save(self, **kwargs):
-        self.clean()
+        should_clean = kwargs.pop("clean", True)
+        if should_clean:
+            self.clean()
         schedule_job = kwargs.pop("schedule_job", True)
         update_fields = kwargs.get("update_fields", None)
         if update_fields is not None:
@@ -407,12 +409,6 @@ def clean_interval_unit(self):
                 code="invalid",
                 params={"queue": self.queue, "interval": config.SCHEDULER_INTERVAL},
             )
-        if self.interval_seconds() <= config.SCHEDULER_INTERVAL:
-            raise ValidationError(
-                _("Job interval is not a multiple of rq_scheduler's interval frequency: %(interval)ss"),
-                code="invalid",
-                params={"interval": config.SCHEDULER_INTERVAL},
-            )
 
     def clean_result_ttl(self) -> None:
         """Throws an error if there are repeats left to run and the result_ttl won't last until the next scheduled time.

scheduler/tests/test_internals.py

@@ -11,7 +11,7 @@
 
 class TestInternals(SchedulerBaseCase):
     def test_get_scheduled_job(self):
-        task = task_factory(TaskType.ONCE, scheduled_time=timezone.now() - timedelta(hours=1))
+        task = task_factory(TaskType.ONCE, scheduled_time=timezone.now() + timedelta(hours=1))
         self.assertEqual(task, get_scheduled_task(TaskType.ONCE, task.id))
         with self.assertRaises(ValueError):
             get_scheduled_task(task.task_type, task.id + 1)

scheduler/tests/test_mgmt_commands/test_import.py

@@ -32,10 +32,10 @@ def test_import__should_schedule_job(self):
         # assert
         self.assertEqual(1, Task.objects.filter(task_type=TaskType.ONCE).count())
         self.assertEqual(1, Task.objects.filter(task_type=TaskType.REPEATABLE).count())
-        tasks = Task.objects.filter(task_type=TaskType.ONCE).first()
+        db_task = Task.objects.filter(task_type=TaskType.ONCE).first()
         attrs = ["name", "queue", "callable", "enabled", "timeout"]
         for attr in attrs:
-            self.assertEqual(getattr(tasks[0], attr), getattr(tasks, attr))
+            self.assertEqual(getattr(tasks[0], attr), getattr(db_task, attr))
 
     def test_import__should_schedule_job_yaml(self):
         tasks = list()

scheduler/tests/test_task_types/test_once_task.py

@@ -35,5 +35,6 @@ def test_create_with_date_in_the_past__fail(self):
         self.assertEqual(str(cm.exception), "{'scheduled_time': ['Scheduled time must be in the future']}")
 
     def test_unschedulable_old_job(self):
-        task = task_factory(self.task_type, scheduled_time=timezone.now() - timedelta(hours=1))
+        task = task_factory(self.task_type, scheduled_time=timezone.now() - timedelta(hours=1), instance_only=True)
+        task.save(clean=False)
         self.assertFalse(task.is_scheduled())

scheduler/tests/test_task_types/test_repeatable_task.py

@@ -13,7 +13,6 @@
 
 
 class TestRepeatableTask(BaseTestCases.TestSchedulableTask):
-    task_type = TaskType.REPEATABLE
     queue_name = settings.get_queue_names()[0]
 
     def test_create_task_error(self):
@@ -35,6 +34,7 @@ def test_create_task_error(self):
         self.assertEqual(task.interval_unit, "seconds")
 
     def test_create_task_without_scheduled_time(self):
+        settings.SCHEDULER_CONFIG.SCHEDULER_INTERVAL = 10
         task = Task.objects.create(
             name="konichiva_every_2s",
             callable="scheduler.tests.jobs.test_args_kwargs",
@@ -52,34 +52,34 @@ def test_create_task_without_scheduled_time(self):
         self.assertEqual(task.interval_unit, "seconds")
 
     def test_unschedulable_old_job(self):
-        task = task_factory(self.task_type, scheduled_time=timezone.now() - timedelta(hours=1), repeat=0)
+        task = task_factory(TaskType.REPEATABLE, scheduled_time=timezone.now() - timedelta(hours=1), repeat=0)
         self.assertFalse(task.is_scheduled())
 
     def test_schedulable_old_job_repeat_none(self):
         # If repeat is None, the job should be scheduled
-        task = task_factory(self.task_type, scheduled_time=timezone.now() - timedelta(hours=1), repeat=None)
+        task = task_factory(TaskType.REPEATABLE, scheduled_time=timezone.now() - timedelta(hours=1), repeat=None)
         self.assertTrue(task.is_scheduled())
 
     def test_clean(self):
-        task = task_factory(self.task_type)
+        task = task_factory(TaskType.REPEATABLE)
         task.queue = self.queue_name
         task.callable = "scheduler.tests.jobs.test_job"
         task.interval = 1
         task.success_ttl = -1
         self.assertIsNone(task.clean())
 
     def test_clean_seconds(self):
-        task = task_factory(self.task_type)
+        task = task_factory(TaskType.REPEATABLE)
         task.queue = self.queue_name
         task.callable = "scheduler.tests.jobs.test_job"
         task.interval = 60
         task.success_ttl = -1
         task.interval_unit = "seconds"
-        self.assertIsNone(task.clean())
+        task.clean()
 
     @override_settings(SCHEDULER_CONFIG=SchedulerConfiguration(SCHEDULER_INTERVAL=10))
     def test_clean_too_frequent(self):
-        task = task_factory(self.task_type)
+        task = task_factory(TaskType.REPEATABLE)
         task.queue = self.queue_name
         task.callable = "scheduler.tests.jobs.test_job"
         task.interval = 2  # Smaller than 10
@@ -89,7 +89,7 @@ def test_clean_too_frequent(self):
             task.clean_interval_unit()
 
     def test_clean_short_result_ttl(self):
-        task = task_factory(self.task_type)
+        task = task_factory(TaskType.REPEATABLE)
         task.queue = self.queue_name
         task.callable = "scheduler.tests.jobs.test_job"
         task.interval = 1
@@ -101,7 +101,7 @@ def test_clean_short_result_ttl(self):
             task.clean_result_ttl()
 
     def test_clean_indefinite_result_ttl(self):
-        task = task_factory(self.task_type)
+        task = task_factory(TaskType.REPEATABLE)
         task.queue = self.queue_name
         task.callable = "scheduler.tests.jobs.test_job"
         task.interval = 1
@@ -110,72 +110,72 @@ def test_clean_indefinite_result_ttl(self):
         task.clean_result_ttl()
 
     def test_clean_undefined_result_ttl(self):
-        task = task_factory(self.task_type)
+        task = task_factory(TaskType.REPEATABLE)
         task.queue = self.queue_name
         task.callable = "scheduler.tests.jobs.test_job"
         task.interval = 1
         task.interval_unit = "hours"
         task.clean_result_ttl()
 
     def test_interval_seconds_weeks(self):
-        task = task_factory(self.task_type, interval=2, interval_unit="weeks")
+        task = task_factory(TaskType.REPEATABLE, interval=2, interval_unit="weeks")
         self.assertEqual(1209600.0, task.interval_seconds())
 
     def test_interval_seconds_days(self):
-        task = task_factory(self.task_type, interval=2, interval_unit="days")
+        task = task_factory(TaskType.REPEATABLE, interval=2, interval_unit="days")
         self.assertEqual(172800.0, task.interval_seconds())
 
     def test_interval_seconds_hours(self):
-        task = task_factory(self.task_type, interval=2, interval_unit="hours")
+        task = task_factory(TaskType.REPEATABLE, interval=2, interval_unit="hours")
         self.assertEqual(7200.0, task.interval_seconds())
 
     def test_interval_seconds_minutes(self):
-        task = task_factory(self.task_type, interval=15, interval_unit="minutes")
+        task = task_factory(TaskType.REPEATABLE, interval=15, interval_unit="minutes")
         self.assertEqual(900.0, task.interval_seconds())
 
     def test_interval_seconds_seconds(self):
-        task = task_factory(self.task_type, interval=15, interval_unit="seconds")
+        task = task_factory(TaskType.REPEATABLE, interval=15, interval_unit="seconds")
         self.assertEqual(15.0, task.interval_seconds())
 
     def test_result_interval(self):
-        task = task_factory(self.task_type)
+        task = task_factory(TaskType.REPEATABLE)
         entry = _get_task_scheduled_job_from_registry(task)
         self.assertEqual(entry.meta["interval"], 3600)
 
     def test_repeat(self):
-        task = task_factory(self.task_type, repeat=10)
+        task = task_factory(TaskType.REPEATABLE, repeat=10)
         entry = _get_task_scheduled_job_from_registry(task)
         self.assertEqual(entry.meta["repeat"], 10)
 
     def test_repeat_old_job_exhausted(self):
         base_time = timezone.now()
-        task = task_factory(self.task_type, scheduled_time=base_time - timedelta(hours=10), repeat=10)
+        task = task_factory(TaskType.REPEATABLE, scheduled_time=base_time - timedelta(hours=10), repeat=10)
         self.assertEqual(task.is_scheduled(), False)
 
     def test_repeat_old_job_last_iter(self):
         base_time = timezone.now()
-        task = task_factory(self.task_type, scheduled_time=base_time - timedelta(hours=9, minutes=30), repeat=10)
+        task = task_factory(TaskType.REPEATABLE, scheduled_time=base_time - timedelta(hours=9, minutes=30), repeat=10)
         self.assertEqual(task.repeat, 0)
         self.assertEqual(task.is_scheduled(), True)
 
     def test_repeat_old_job_remaining(self):
         base_time = timezone.now()
-        task = task_factory(self.task_type, scheduled_time=base_time - timedelta(minutes=30), repeat=5)
+        task = task_factory(TaskType.REPEATABLE, scheduled_time=base_time - timedelta(minutes=30), repeat=5)
         self.assertEqual(task.repeat, 4)
         self.assertEqual(task.scheduled_time, base_time + timedelta(minutes=30))
         self.assertEqual(task.is_scheduled(), True)
 
     def test_repeat_none_interval_2_min(self):
         base_time = timezone.now()
-        task = task_factory(self.task_type, scheduled_time=base_time - timedelta(minutes=29), repeat=None)
+        task = task_factory(TaskType.REPEATABLE, scheduled_time=base_time - timedelta(minutes=29), repeat=None)
         task.interval = 120
         task.interval_unit = "seconds"
         task.save()
         self.assertTrue(task.scheduled_time > base_time)
         self.assertTrue(task.is_scheduled())
 
     def test_check_rescheduled_after_execution(self):
-        task = task_factory(self.task_type, scheduled_time=timezone.now() + timedelta(seconds=1), repeat=10)
+        task = task_factory(TaskType.REPEATABLE, scheduled_time=timezone.now() + timedelta(seconds=1), repeat=10)
         queue = task.rqueue
         first_run_id = task.job_name
         entry = JobModel.get(first_run_id, connection=queue.connection)
@@ -190,7 +190,7 @@ def test_check_rescheduled_after_execution(self):
 
     def test_check_rescheduled_after_execution_failed_job(self):
         task = task_factory(
-            self.task_type,
+            TaskType.REPEATABLE,
             callable_name="scheduler.tests.jobs.failing_job",
             scheduled_time=timezone.now() + timedelta(seconds=1),
             repeat=10,
@@ -209,7 +209,7 @@ def test_check_rescheduled_after_execution_failed_job(self):
 
     def test_check_not_rescheduled_after_last_repeat(self):
         task = task_factory(
-            self.task_type,
+            TaskType.REPEATABLE,
             scheduled_time=timezone.now() + timedelta(seconds=1),
             repeat=1,
         )