Return None if run_in_worker doesn't create a new JobRequest, fix typing issues and intentional delay 0 on job retry

Author: davegaeddert

plain-jobs/plain/jobs/jobs.py

@@ -62,7 +62,7 @@ def run_in_worker(
         retries: int | None = None,
         retry_attempt: int = 0,
         unique_key: str | None = None,
-    ) -> JobRequest | list[JobRequest | JobProcess]:
+    ) -> JobRequest | None:
         from .models import JobRequest
 
         job_class_name = jobs_registry.get_job_class_name(self.__class__)
@@ -170,8 +170,7 @@ def run_in_worker(
                 span.set_attribute(ERROR_TYPE, "IntegrityError")
                 span.set_status(trace.Status(trace.StatusCode.ERROR, "Duplicate job"))
                 logger.warning("Job already in progress: %s", e)
-                # Try to return the _in_progress list again
-                return self._in_progress(unique_key)
+                return None
 
     def _in_progress(self, unique_key: str) -> list[JobRequest | JobProcess]:
         """Get all JobRequests and JobProcess that are currently in progress, regardless of queue."""

plain-jobs/plain/jobs/models.py

@@ -412,10 +412,14 @@ class JobResult(models.Model):
         ],
     )
 
-    def retry_job(self, delay: int | None = None) -> JobRequest:
+    def retry_job(self, delay: int | None = None) -> JobRequest | None:
         retry_attempt = self.retry_attempt + 1
         job = jobs_registry.load_job(self.job_class, self.parameters)
-        retry_delay = delay or job.get_retry_delay(retry_attempt)
+
+        if delay is None:
+            retry_delay = job.get_retry_delay(retry_attempt)
+        else:
+            retry_delay = delay
 
         with transaction.atomic():
             result = job.run_in_worker(
@@ -427,12 +431,9 @@ def retry_job(self, delay: int | None = None) -> JobRequest:
                 retry_attempt=retry_attempt,
                 # Unique key could be passed also?
             )
+            if result:
+                self.retry_job_request_uuid = result.uuid
+                self.save(update_fields=["retry_job_request_uuid"])
+                return result
 
-            # TODO it is actually possible that result is a list
-            # of pending jobs, which would need to be handled...
-            # Right now it will throw an exception which could be caught by retry_failed_jobs.
-
-            self.retry_job_request_uuid = result.uuid  # type: ignore
-            self.save(update_fields=["retry_job_request_uuid"])
-
-        return result  # type: ignore
+        return None