Merge branch 'master' into mypy

# Conflicts:
#	scheduler/models/task.py
#	scheduler/tests/test_task_types/test_repeatable_task.py
#	uv.lock

Author: cunla

docs/changelog.md

@@ -1,5 +1,17 @@
 # Changelog
 
+## v4.0.5 🌈
+
+### 🐛 Bug Fixes
+
+- fix:repeatable task without start date #276
+- fix:admin list of tasks showing local datetime #280
+- fix:wait for job child process using os.waitpid #281
+
+### 🧰 Maintenance
+
+- refactor some tests
+
 ## v4.0.4 🌈
 
 ### 🐛 Bug Fixes

pyproject.toml

@@ -4,10 +4,10 @@ build-backend = "hatchling.build"
 
 [project]
 name = "django-tasks-scheduler"
-version = "4.0.4"
+version = "4.0.5"
 description = "An async job scheduler for django using redis/valkey brokers"
 authors = [{ name = "Daniel Moran", email = "[email protected]" }]
-requires-python = "~=3.10"
+requires-python = ">=3.10"
 readme = "README.md"
 license = "MIT"
 maintainers = [{ name = "Daniel Moran", email = "[email protected]" }]

scheduler/admin/task_admin.py

@@ -4,6 +4,7 @@
 from django.contrib.contenttypes.admin import GenericStackedInline
 from django.db.models import QuerySet
 from django.http import HttpRequest
+from django.utils import timezone, formats
 from django.utils.translation import gettext_lazy as _
 
 from scheduler.helpers.queues import get_queue
@@ -120,7 +121,11 @@ class Media:
     @admin.display(description="Schedule")
     def task_schedule(self, o: Task) -> str:
         if o.task_type == TaskType.ONCE.value:
-            return f"Run once: {o.scheduled_time:%Y-%m-%d %H:%M:%S}"
+            if timezone.is_naive(o.scheduled_time):
+                local_time = timezone.make_aware(o.scheduled_time, timezone.get_current_timezone())
+            else:
+                local_time = timezone.localtime(o.scheduled_time)
+            return f"Run once: {formats.date_format(local_time, 'DATETIME_FORMAT')}"
         elif o.task_type == TaskType.CRON.value:
             return f"Cron: {o.cron_string}"
         else:  # if o.task_type == TaskType.REPEATABLE.value:

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():
@@ -187,9 +187,9 @@ def is_scheduled(self) -> bool:
             return False
         # check whether job_id is in scheduled/queued/active jobs
         res = (
-                (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())
+            (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)
@@ -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

@@ -21,21 +21,21 @@ def tearDown(self) -> None:
         os.remove(self.tmpfile.name)
 
     def test_import__should_schedule_job(self):
-        jobs = list()
-        jobs.append(task_factory(TaskType.ONCE, enabled=True, instance_only=True))
-        jobs.append(task_factory(TaskType.REPEATABLE, enabled=True, instance_only=True))
-        res = json.dumps([j.to_dict() for j in jobs])
+        tasks = list()
+        tasks.append(task_factory(TaskType.ONCE, enabled=True, instance_only=True))
+        tasks.append(task_factory(TaskType.REPEATABLE, enabled=True, instance_only=True))
+        res = json.dumps([j.to_dict() for j in tasks])
         self.tmpfile.write(res)
         self.tmpfile.flush()
         # act
         call_command("import", filename=self.tmpfile.name)
         # assert
         self.assertEqual(1, Task.objects.filter(task_type=TaskType.ONCE).count())
         self.assertEqual(1, Task.objects.filter(task_type=TaskType.REPEATABLE).count())
-        db_job = 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(jobs[0], attr), getattr(db_job, attr))
+            self.assertEqual(getattr(tasks[0], attr), getattr(db_task, attr))
 
     def test_import__should_schedule_job_yaml(self):
         tasks = list()
@@ -49,16 +49,16 @@ def test_import__should_schedule_job_yaml(self):
         # assert
         self.assertEqual(1, Task.objects.filter(task_type=TaskType.ONCE).count())
         self.assertEqual(1, Task.objects.filter(task_type=TaskType.REPEATABLE).count())
-        db_job = Task.objects.filter(task_type=TaskType.ONCE).first()
+        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(db_job, attr))
+            self.assertEqual(getattr(tasks[0], attr), getattr(task, attr))
 
     def test_import__should_schedule_job_yaml_without_yaml_lib(self):
-        jobs = list()
-        jobs.append(task_factory(TaskType.ONCE, enabled=True, instance_only=True))
-        jobs.append(task_factory(TaskType.REPEATABLE, enabled=True, instance_only=True))
-        res = yaml.dump([j.to_dict() for j in jobs], default_flow_style=False)
+        tasks = list()
+        tasks.append(task_factory(TaskType.ONCE, enabled=True, instance_only=True))
+        tasks.append(task_factory(TaskType.REPEATABLE, enabled=True, instance_only=True))
+        res = yaml.dump([j.to_dict() for j in tasks], default_flow_style=False)
         self.tmpfile.write(res)
         self.tmpfile.flush()
         # act
@@ -68,27 +68,27 @@ def test_import__should_schedule_job_yaml_without_yaml_lib(self):
             self.assertEqual(cm.exception.code, 1)
 
     def test_import__should_schedule_job_reset(self):
-        jobs = list()
+        tasks = list()
         task_factory(TaskType.ONCE, enabled=True)
         task_factory(TaskType.ONCE, enabled=True)
-        jobs.append(task_factory(TaskType.ONCE, enabled=True))
-        jobs.append(task_factory(TaskType.REPEATABLE, enabled=True, instance_only=True))
-        res = json.dumps([j.to_dict() for j in jobs])
+        tasks.append(task_factory(TaskType.ONCE, enabled=True))
+        tasks.append(task_factory(TaskType.REPEATABLE, enabled=True, instance_only=True))
+        res = json.dumps([j.to_dict() for j in tasks])
         self.tmpfile.write(res)
         self.tmpfile.flush()
         # act
         call_command("import", filename=self.tmpfile.name, reset=True)
         # assert
         self.assertEqual(1, Task.objects.filter(task_type=TaskType.ONCE).count())
-        db_job = Task.objects.filter(task_type=TaskType.ONCE).first()
+        task = Task.objects.filter(task_type=TaskType.ONCE).first()
         attrs = ["name", "queue", "callable", "enabled", "timeout"]
         for attr in attrs:
-            self.assertEqual(getattr(jobs[0], attr), getattr(db_job, attr))
+            self.assertEqual(getattr(tasks[0], attr), getattr(task, attr))
         self.assertEqual(1, Task.objects.filter(task_type=TaskType.REPEATABLE).count())
-        db_job = Task.objects.filter(task_type=TaskType.REPEATABLE).first()
+        task = Task.objects.filter(task_type=TaskType.REPEATABLE).first()
         attrs = ["name", "queue", "callable", "enabled", "timeout"]
         for attr in attrs:
-            self.assertEqual(getattr(jobs[1], attr), getattr(db_job, attr))
+            self.assertEqual(getattr(tasks[1], attr), getattr(task, attr))
 
     def test_import__should_schedule_job_update_existing(self):
         tasks = list()
@@ -101,10 +101,10 @@ def test_import__should_schedule_job_update_existing(self):
         call_command("import", filename=self.tmpfile.name, update=True)
         # assert
         self.assertEqual(2, Task.objects.filter(task_type=TaskType.ONCE).count())
-        db_job = Task.objects.filter(task_type=TaskType.ONCE).get(name=tasks[0].name)
+        task = Task.objects.filter(task_type=TaskType.ONCE).get(name=tasks[0].name)
         attrs = ["name", "queue", "callable", "enabled", "timeout"]
         for attr in attrs:
-            self.assertEqual(getattr(tasks[0], attr), getattr(db_job, attr))
+            self.assertEqual(getattr(tasks[0], attr), getattr(task, attr))
 
     def test_import__should_schedule_job_without_update_existing(self):
         tasks = list()
@@ -117,7 +117,7 @@ def test_import__should_schedule_job_without_update_existing(self):
         call_command("import", filename=self.tmpfile.name)
         # assert
         self.assertEqual(2, Task.objects.filter(task_type=TaskType.ONCE).count())
-        db_job = Task.objects.get(name=tasks[0].name)
+        task = Task.objects.get(name=tasks[0].name)
         attrs = ["id", "name", "queue", "callable", "enabled", "timeout"]
         for attr in attrs:
-            self.assertEqual(getattr(tasks[0], attr), getattr(db_job, attr))
+            self.assertEqual(getattr(tasks[0], attr), getattr(task, attr))

scheduler/tests/test_task_types/test_once_task.py

@@ -1,6 +1,8 @@
 from datetime import timedelta, datetime
 
+import time_machine
 from django.core.exceptions import ValidationError
+from django.urls import reverse
 from django.utils import timezone
 
 from scheduler import settings
@@ -9,15 +11,26 @@
 from scheduler.tests.testtools import task_factory
 
 
-class TestScheduledTask(BaseTestCases.TestSchedulableTask):
+class TestScheduledOnceTask(BaseTestCases.TestSchedulableTask):
     task_type = TaskType.ONCE
     queue_name = settings.get_queue_names()[0]
 
     def test_clean(self):
-        job = task_factory(self.task_type)
-        job.queue = self.queue_name
-        job.callable = "scheduler.tests.jobs.test_job"
-        self.assertIsNone(job.clean())
+        task = task_factory(self.task_type)
+        task.queue = self.queue_name
+        task.callable = "scheduler.tests.jobs.test_job"
+        self.assertIsNone(task.clean())
+
+    @time_machine.travel(datetime(2016, 12, 25))
+    def test_admin_changelist_view__has_timezone_data(self):
+        # arrange
+        self.client.login(username="admin", password="admin")
+        task_factory(self.task_type)
+        url = reverse("admin:scheduler_task_changelist")
+        # act
+        res = self.client.get(url)
+        # assert
+        self.assertContains(res, "Run once: Dec. 26, 2016, midnight", count=1, status_code=200)
 
     def test_create_without_date__fail(self):
         task = task_factory(self.task_type, scheduled_time=None, instance_only=True)
@@ -35,5 +48,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):
-        job = task_factory(self.task_type, scheduled_time=timezone.now() - timedelta(hours=1))
-        self.assertFalse(job.is_scheduled())
+        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())