Freeze Task class

Author: RealOrangeOne

django_tasks/task.py

@@ -1,5 +1,4 @@
-from copy import deepcopy
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, replace
 from datetime import datetime, timedelta
 from inspect import iscoroutinefunction
 from typing import (
@@ -53,7 +52,7 @@ class ResultStatus(TextChoices):
 P = ParamSpec("P")
 
 
-@dataclass
+@dataclass(frozen=True)
 class Task(Generic[P, T]):
     priority: int
     """The priority of the task"""
@@ -98,22 +97,21 @@ def using(
         Create a new task with modified defaults
         """
 
-        task = deepcopy(self)
+        changes: Dict[str, Any] = {}
+
         if priority is not None:
-            task.priority = priority
+            changes["priority"] = priority
         if queue_name is not None:
-            task.queue_name = queue_name
+            changes["queue_name"] = queue_name
         if run_after is not None:
             if isinstance(run_after, timedelta):
-                task.run_after = timezone.now() + run_after
+                changes["run_after"] = timezone.now() + run_after
             else:
-                task.run_after = run_after
+                changes["run_after"] = run_after
         if backend is not None:
-            task.backend = backend
-
-        task.get_backend().validate_task(task)
+            changes["backend"] = backend
 
-        return task
+        return replace(self, **changes)
 
     def enqueue(self, *args: P.args, **kwargs: P.kwargs) -> "TaskResult[T]":
         """

tests/tests/test_tasks.py

@@ -113,6 +113,18 @@ def test_using_creates_new_instance(self) -> None:
         self.assertEqual(new_task, test_tasks.noop_task)
         self.assertIsNot(new_task, test_tasks.noop_task)
 
+    def test_chained_using(self) -> None:
+        now = timezone.now()
+
+        run_after_task = test_tasks.noop_task.using(run_after=now)
+        self.assertEqual(run_after_task.run_after, now)
+
+        priority_task = run_after_task.using(priority=10)
+        self.assertEqual(priority_task.priority, 10)
+        self.assertEqual(priority_task.run_after, now)
+
+        self.assertEqual(run_after_task.priority, 0)
+
     async def test_refresh_result(self) -> None:
         result = await test_tasks.noop_task.aenqueue()