Tweaks

Author: RealOrangeOne

django_tasks/backends/database/apps.py

@@ -5,3 +5,6 @@ class TasksAppConfig(AppConfig):
     name = "django_tasks.backends.database"
     label = "django_tasks_database"
     verbose_name = "Tasks Database Backend"
+
+    def ready(self) -> None:
+        from . import signal_handlers  # noqa

django_tasks/backends/database/backend.py

@@ -44,7 +44,7 @@ def _task_to_db_task(
             priority=task.priority,
             task_path=task.module_path,
             queue_name=task.queue_name,
-            run_after=task.run_after,
+            run_after=task.run_after,  # type: ignore[misc]
             backend_name=self.alias,
         )
 

django_tasks/backends/database/migrations/0016_alter_dbtaskresult_options_and_more.py

@@ -1,45 +1,70 @@
-# Generated by Django 5.1.4 on 2025-04-28 11:28
+# Generated by Django 5.1.5 on 2025-05-02 12:05
 
 import datetime
+
 from django.db import migrations, models
 
 
 class Migration(migrations.Migration):
-
     dependencies = [
-        ('django_tasks_database', '0015_correctly_order_run_after'),
+        ("django_tasks_database", "0015_correctly_order_run_after"),
     ]
 
     operations = [
         migrations.AlterModelOptions(
-            name='dbtaskresult',
-            options={'ordering': [models.OrderBy(models.F('priority'), descending=True), models.OrderBy(models.F('run_after'))], 'verbose_name': 'Task Result', 'verbose_name_plural': 'Task Results'},
+            name="dbtaskresult",
+            options={
+                "ordering": [
+                    models.OrderBy(models.F("priority"), descending=True),
+                    models.OrderBy(models.F("run_after")),
+                ],
+                "verbose_name": "Task Result",
+                "verbose_name_plural": "Task Results",
+            },
         ),
         migrations.AlterField(
-            model_name='dbtaskresult',
-            name='backend_name',
-            field=models.CharField(max_length=32, verbose_name='backend name'),
+            model_name="dbtaskresult",
+            name="backend_name",
+            field=models.CharField(max_length=32, verbose_name="backend name"),
         ),
         migrations.AlterField(
-            model_name='dbtaskresult',
-            name='queue_name',
-            field=models.CharField(default='default', max_length=32, verbose_name='queue name'),
+            model_name="dbtaskresult",
+            name="queue_name",
+            field=models.CharField(
+                default="default", max_length=32, verbose_name="queue name"
+            ),
         ),
         migrations.AlterField(
-            model_name='dbtaskresult',
-            name='run_after',
-            field=models.DateTimeField(default=datetime.datetime(9999, 1, 1, 0, 0, tzinfo=datetime.timezone.utc), verbose_name='run after'),
+            model_name="dbtaskresult",
+            name="run_after",
+            field=models.DateTimeField(
+                default=datetime.datetime(
+                    9999, 12, 2, 5, 59, 59, 999999, tzinfo=datetime.timezone.utc
+                ),
+                verbose_name="run after",
+            ),
+            preserve_default=False,
         ),
         migrations.AddIndex(
-            model_name='dbtaskresult',
-            index=models.Index(models.F('status'), models.OrderBy(models.F('priority'), descending=True), models.OrderBy(models.F('run_after')), condition=models.Q(('status', 'NEW')), name='django_task_new_ordering_idx'),
+            model_name="dbtaskresult",
+            index=models.Index(
+                models.F("status"),
+                models.OrderBy(models.F("priority"), descending=True),
+                models.OrderBy(models.F("run_after")),
+                condition=models.Q(("status", "NEW")),
+                name="django_task_new_ordering_idx",
+            ),
         ),
         migrations.AddIndex(
-            model_name='dbtaskresult',
-            index=models.Index(fields=['queue_name'], name='django_task_queue_n_99c321_idx'),
+            model_name="dbtaskresult",
+            index=models.Index(
+                fields=["queue_name"], name="django_task_queue_n_99c321_idx"
+            ),
         ),
         migrations.AddIndex(
-            model_name='dbtaskresult',
-            index=models.Index(fields=['backend_name'], name='django_task_backend_071754_idx'),
+            model_name="dbtaskresult",
+            index=models.Index(
+                fields=["backend_name"], name="django_task_backend_071754_idx"
+            ),
         ),
     ]

django_tasks/backends/database/models.py

@@ -47,7 +47,7 @@ def __class_getitem__(cls, _):
             return cls
 
 
-DATE_MAX = datetime.datetime(9999, 1, 1, tzinfo=datetime.timezone.utc)
+DATE_MAX = datetime.datetime.max.replace(day=1).astimezone(datetime.timezone.utc)
 
 
 class DBTaskResultQuerySet(models.QuerySet):
@@ -104,7 +104,7 @@ class DBTaskResult(GenericBase[P, T], models.Model):
     )
     backend_name = models.CharField(_("backend name"), max_length=32)
 
-    run_after = models.DateTimeField(_("run after"), default=DATE_MAX)
+    run_after = models.DateTimeField(_("run after"))
 
     return_value = models.JSONField(_("return value"), default=None, null=True)
 
@@ -119,7 +119,8 @@ class Meta:
         verbose_name_plural = _("Task Results")
         indexes = [
             models.Index(
-                "status", *ordering,
+                "status",
+                *ordering,
                 name="django_task_new_ordering_idx",
                 condition=Q(status=ResultStatus.NEW),
             ),
@@ -142,11 +143,6 @@ class Meta:
                 )
             ]
 
-    def save(self, **kwargs):
-        if self.run_after is None:
-            self.run_after = DATE_MAX
-        super().save(**kwargs)
-
     @property
     def task(self) -> Task[P, T]:
         task = import_string(self.task_path)
@@ -156,13 +152,10 @@ def task(self) -> Task[P, T]:
                 f"Task {self.id} does not point to a Task ({self.task_path})"
             )
 
-        if self.run_after == DATE_MAX:
-            self.run_after = None
-
         return task.using(
             priority=self.priority,
             queue_name=self.queue_name,
-            run_after=self.run_after,
+            run_after=None if self.run_after == DATE_MAX else self.run_after,
             backend=self.backend_name,
         )
 

django_tasks/backends/database/signal_handlers.py

@@ -0,0 +1,12 @@
+from typing import Any
+
+from django.db.models.signals import pre_save
+from django.dispatch import receiver
+
+from .models import DATE_MAX, DBTaskResult
+
+
+@receiver(pre_save, sender=DBTaskResult)
+def set_run_after(sender: Any, instance: DBTaskResult, **kwargs: Any) -> None:
+    if instance.run_after is None:
+        instance.run_after = DATE_MAX

tests/tests/test_database_backend.py

@@ -1,3 +1,4 @@
+import copy
 import json
 import logging
 import os
@@ -389,15 +390,11 @@ def test_enqueue_logs(self) -> None:
 
     def test_index_scan_for_ready(self) -> None:
         test_tasks.noop_task.enqueue()
-        db_task = DBTaskResult.objects.first()
-        new_db_tasks = []
-        for _ in range(5000):
-            new_db_task = DBTaskResult()
-            for f in DBTaskResult._meta.fields:
-                if f.name != "id":
-                    setattr(new_db_task, f.attname, getattr(db_task, f.attname))
-            new_db_tasks.append(new_db_task)
-        DBTaskResult.objects.bulk_create(new_db_tasks)
+
+        # Quickly duplicate tasks
+        db_task = DBTaskResult.objects.get()
+        db_task.id = None
+        DBTaskResult.objects.bulk_create([copy.copy(db_task) for _ in range(5000)])
 
         # Update query plan for certain databases
         if connection.vendor == "postgresql":