fix error with run_after enqueueing

Author: knifecake

README.md

@@ -4,8 +4,7 @@ Steady Queue is a port to Django of the excellent [Solid
 Queue][solid-queue-github] DB-based queueing backend for Ruby on Rails. The goal
 of this port has been to keep the internals as close as possible to a direct
 translation from Ruby to Python, while adapting the external interfaces to be
-idiomatic in Django and Python. Read more on the differences between Steady
-Queue and Solid Queue below.
+idiomatic in Django. Read more on the [differences between Steady Queue and Solid Queue below](#deviations-from-solid-queue).
 
 Steady queue exposes a task backend that is compatible with the background worker specification outlined in [DEP 0014][DEP0014]. In addition to task enqueueing and processing, it supports delayed tasks, concurrency controls, recurring tasks, pausing queues, numeric priorities per task, priorities by queue order and bulk enqueueing.
 
@@ -39,7 +38,7 @@ Steady Queue works like any other DEP 0014-compatible task backend.
 Tasks are functions decorated with the `@task` decorator from `django.tasks`:
 
 ```python
-from django.task import task
+from django.tasks import task
 
 @task()
 def greet(name: str, times: int = 1):

steady_queue/backend.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 from django.tasks import Task, TaskResult, TaskResultStatus
 from django.tasks.backends.base import BaseTaskBackend
 
@@ -17,23 +19,25 @@ def validate_task(self, task: Task) -> None:
         # TODO: do we need to do anything here?
         super().validate_task(task)
 
-    def enqueue(self, task, args, kwargs) -> TaskResult:
+    def enqueue(
+        self, task: SteadyQueueTask, args: list, kwargs: dict[str, Any]
+    ) -> TaskResult:
         from steady_queue.models import Job
 
         if not isinstance(task, SteadyQueueTask):
             raise ValueError("Steady Queue only supports SteadyQueueTasks")
 
-        task.args = args
-        task.kwargs = kwargs
-        job = Job.objects.enqueue(task, scheduled_at=task.run_after)
-        return self._to_task_result(task, job)
+        job = Job.objects.enqueue(task, args, kwargs)
+        return self._to_task_result(task, job, args, kwargs)
 
     def get_result(self, result_id: str) -> TaskResult:
         raise NotImplementedError(
             "This backend does not support retrieving or refreshing results."
         )
 
-    def _to_task_result(self, task: SteadyQueueTask, job) -> TaskResult:
+    def _to_task_result(
+        self, task: SteadyQueueTask, job, args: list, kwargs: dict[str, Any]
+    ) -> TaskResult:
         return TaskResult(
             task=task,
             id=str(job.id),
@@ -42,8 +46,8 @@ def _to_task_result(self, task: SteadyQueueTask, job) -> TaskResult:
             started_at=None,
             finished_at=job.finished_at,
             last_attempted_at=None,
-            args=[],
-            kwargs=task.arguments,
+            args=args,
+            kwargs=kwargs,
             backend=task.backend,
             errors=[],
             worker_ids=[],

steady_queue/models/job.py

@@ -1,4 +1,3 @@
-from datetime import datetime
 from typing import Optional
 
 from django.db import models
@@ -10,11 +9,11 @@
 
 
 class JobQuerySet(ExecutableQuerySet, models.QuerySet):
-    def enqueue(self, task: SteadyQueueTask, scheduled_at: Optional[datetime] = None):
+    def enqueue(self, task: SteadyQueueTask, args: list, kwargs: dict):
         try:
-            enqueued_job = self.create(**self.model.attributes_from_django_task(task))
-            task.provider_task_id = enqueued_job.id
-            return enqueued_job
+            return self.create(
+                **self.model.attributes_from_django_task(task, args, kwargs)
+            )
         except Exception as e:
             # TODO: enqueue error
             raise e
@@ -64,14 +63,16 @@ class Meta:
     DEFAULT_PRIORITY = 0
 
     @classmethod
-    def attributes_from_django_task(cls, task: SteadyQueueTask):
+    def attributes_from_django_task(
+        cls, task: SteadyQueueTask, args: list, kwargs: dict
+    ):
         return {
             "queue_name": task.queue_name or cls.DEFAULT_QUEUE_NAME,
             "django_task_id": task.id,
             "priority": task.priority or cls.DEFAULT_PRIORITY,
             "scheduled_at": task.run_after or timezone.now(),
             "class_name": task.module_path,
-            "arguments": task.serialize(),
+            "arguments": task.serialize(args, kwargs),
             "concurrency_key": task.concurrency_key,
         }
 

steady_queue/recurring_task.py

@@ -1,5 +1,4 @@
 import logging
-from dataclasses import replace
 from typing import Optional
 
 from steady_queue.configuration import Configuration
@@ -41,13 +40,13 @@ def wrapper(task: SteadyQueueTask):
             key=key,
             class_name=class_name,
             schedule=schedule,
-            arguments=task.serialize(),
+            arguments=task.serialize(args, kwargs),
             queue_name=queue_name,
             priority=priority,
             description=description,
         )
         configurations.append(configuration)
 
-        return replace(task, args=args, kwargs=kwargs)
+        return task
 
     return wrapper

steady_queue/task.py

@@ -1,6 +1,6 @@
 import datetime
 import uuid
-from dataclasses import dataclass, replace
+from dataclasses import dataclass
 from typing import Any, Optional
 
 from django.tasks import Task, TaskResult
@@ -16,9 +16,6 @@ class UnknownTaskClassError(Exception):
 
 @dataclass(frozen=True, slots=True, kw_only=True)
 class SteadyQueueTask(Task):
-    args: Optional[list[Any]] = None
-    kwargs: Optional[dict[str, Any]] = None
-
     concurrency_key: Optional[str] = None
     concurrency_limit: Optional[int] = None
     concurrency_duration: Optional[timezone.timedelta] = None
@@ -42,7 +39,7 @@ def using(
         if isinstance(run_after, datetime.timedelta):
             run_after = timezone.now() + run_after
 
-        return super().using(
+        return super(SteadyQueueTask, self).using(
             priority=priority,
             queue_name=queue_name,
             run_after=run_after,
@@ -52,14 +49,14 @@ def using(
     def enqueue(self, *args: Any, **kwargs: Any) -> TaskResult:
         return self.get_backend().enqueue(self, args, kwargs)
 
-    def serialize(self):
+    def serialize(self, args: list, kwargs: dict):
         return {
             "class_name": self.module_path,
             "job_id": self.id,  # TODO: make it stable
             "backend": self.backend,
             "queue_name": self.queue_name,
             "priority": self.priority,
-            "arguments": Arguments.serialize_args_and_kwargs(self.args, self.kwargs),
+            "arguments": Arguments.serialize_args_and_kwargs(args, kwargs),
             "locale": translation.get_language(),
             "timezone": timezone.get_current_timezone_name(),
             "enqueued_at": timezone.now().isoformat(),
@@ -68,8 +65,9 @@ def serialize(self):
 
     @classmethod
     def execute(cls, job_data: dict[str, Any]):
+        args, kwargs = Arguments.deserialize_args_and_kwargs(job_data["arguments"])
         task = cls.deserialize(job_data)
-        task.func(*task.args, **task.kwargs)
+        task.func(*args, **kwargs)
 
     @classmethod
     def deserialize(cls, job_data: dict[str, Any]):
@@ -78,11 +76,13 @@ def deserialize(cls, job_data: dict[str, Any]):
         except ImportError as e:
             raise UnknownTaskClassError(job_data["class_name"]) from e
 
-        task = task_class.using(
+        return task_class.using(
             priority=job_data["priority"],
             queue_name=job_data["queue_name"],
-            run_after=job_data["scheduled_at"],
+            run_after=(
+                timezone.datetime.fromisoformat(job_data["scheduled_at"])
+                if job_data["scheduled_at"]
+                else None
+            ),
             backend=job_data["backend"],
         )
-        args, kwargs = Arguments.deserialize_args_and_kwargs(job_data["arguments"])
-        return replace(task, args=args, kwargs=kwargs)