jobs: updated last_success condition and update_run

Author: 0einstein0

invenio_jobs/jobs.py

@@ -101,7 +101,10 @@ def _build_task_arguments(cls, job_obj, since=None, custom_args=None, **kwargs):
         if custom_args:
             return custom_args
 
-        if since is None and job_obj.last_runs["success"]:
+        last_success = job_obj.last_runs.get("success")
+
+        if since is None and last_success and last_success.started_at:
+
             """
             The most common case: `since` has not been manually specified by the user, so we
             set it to the start time of the last successful job.
@@ -110,7 +113,7 @@ def _build_task_arguments(cls, job_obj, since=None, custom_args=None, **kwargs):
             For comparison with other dates in job implementors, it's useful to have TZ info in the timestamp.
             """
 
-            since = job_obj.last_runs["success"].started_at.replace(tzinfo=timezone.utc)
+            since = last_success.started_at.replace(tzinfo=timezone.utc)
 
         """
         Otherwise, since is already specified as a datetime with a timezone (see PredefinedArgsSchema) or we have never

invenio_jobs/tasks.py

@@ -27,23 +27,32 @@ def update_run(run, **kwargs):
     """Method to update and commit run updates."""
     if not run:
         return
+
     has_active_subtasks = (
         run.subtasks.filter(
             Run.status.in_([RunStatusEnum.RUNNING.value, RunStatusEnum.QUEUED.value])
         ).count()
         > 0
     )
+
     current_app.logger.info(
         f"Updating run {run.id} with status {run.status} and active subtasks: {has_active_subtasks}"
     )
-    if has_active_subtasks:
-        # If there are active subtasks, we keep the run status as RUNNING and simply update the errored entries, if present.
-        if errored_entries := kwargs.get("errored_entries", None):
+
+    new_status = kwargs.get("status")
+    if has_active_subtasks and new_status != RunStatusEnum.RUNNING:
+        # If subtasks are active, only update errored_entries
+        if errored_entries := kwargs.get("errored_entries"):
             run.errored_entries += errored_entries
-            db.session.commit()
+        db.session.commit()
         return
+
+    # Update all fields (either no active subtasks, or setting status to RUNNING)
     for kw, value in kwargs.items():
-        setattr(run, kw, value)
+        if kw == "errored_entries":
+            run.errored_entries += value
+        else:
+            setattr(run, kw, value)
     db.session.commit()