Modified workflow entry points to return a dictionary so that we can control message requeueing logic as well

Author: tieneupin

src/murfey/server/feedback.py

@@ -2462,17 +2462,17 @@ def feedback_callback(header: dict, message: dict, _db=murfey_db) -> None:
                 return None
             # Run the workflow if a match is found
             workflow: EntryPoint = workflows[0]
-            result = workflow.load()(
+            result: dict[str, bool] = workflow.load()(
                 message=message,
                 murfey_db=_db,
             )
             if murfey.server._transport_object:
-                if result:
+                if result.get("success", False):
                     murfey.server._transport_object.transport.ack(header)
                 else:
                     # Send it directly to DLQ without trying to rerun it
                     murfey.server._transport_object.transport.nack(
-                        header, requeue=False
+                        header, requeue=result.get("requeue", False)
                     )
             if not result:
                 logger.error(

src/murfey/workflows/clem/align_and_merge.py

@@ -78,4 +78,4 @@ def submit_cluster_request(
         },
         new_connection=True,
     )
-    return True
+    return {"success": True}

src/murfey/workflows/clem/register_align_and_merge_results.py

@@ -41,7 +41,7 @@ def parse_stringified_list(cls, value):
 
 def register_align_and_merge_result(
     message: dict, murfey_db: Session, demo: bool = False
-) -> bool:
+) -> dict[str, bool]:
     """
     session_id (recipe)
     register (wrapper)
@@ -69,13 +69,13 @@ def register_align_and_merge_result(
                 "Invalid type for align-and-merge processing result: "
                 f"{type(message['result'])}"
             )
-            return False
+            return {"success": False, "requeue": False}
     except Exception:
         logger.error(
             "Exception encountered when parsing align-and-merge processing result: \n"
             f"{traceback.format_exc()}"
         )
-        return False
+        return {"success": False, "requeue": False}
 
     # Outer try-finally block for tidying up database-related section of function
     try:
@@ -103,8 +103,8 @@ def register_align_and_merge_result(
                 f"{result.series_name!r}: \n"
                 f"{traceback.format_exc()}"
             )
-            return False
+            return {"success": False, "requeue": False}
 
-        return True
+        return {"success": True}
     finally:
         murfey_db.close()

src/murfey/workflows/clem/register_preprocessing_results.py

@@ -56,7 +56,7 @@ class CLEMPreprocessingResult(BaseModel):
     extent: list[float]
 
 
-def run(message: dict, murfey_db: Session, demo: bool = False) -> bool:
+def run(message: dict, murfey_db: Session, demo: bool = False) -> dict[str, bool]:
     session_id: int = (
         int(message["session_id"])
         if not isinstance(message["session_id"], int)
@@ -72,13 +72,13 @@ def run(message: dict, murfey_db: Session, demo: bool = False) -> bool:
             logger.error(
                 f"Invalid type for TIFF preprocessing result: {type(message['result'])}"
             )
-            return False
+            return {"success": False, "requeue": False}
     except Exception:
         logger.error(
             "Exception encountered when parsing TIFF preprocessing result: \n"
             f"{traceback.format_exc()}"
         )
-        return False
+        return {"success": False, "requeue": False}
 
     # Outer try-finally block for tidying up database-related section of function
     try:
@@ -181,7 +181,7 @@ def run(message: dict, murfey_db: Session, demo: bool = False) -> bool:
                 f"{result.series_name!r}: \n"
                 f"{traceback.format_exc()}"
             )
-            return False
+            return {"success": False, "requeue": False}
 
         # Load instrument name
         try:
@@ -197,7 +197,7 @@ def run(message: dict, murfey_db: Session, demo: bool = False) -> bool:
                 f"Error requesting data from database for {result.series_name!r} series: \n"
                 f"{traceback.format_exc()}"
             )
-            return False
+            return {"success": False, "requeue": False}
 
         # Construct list of files to use for image alignment and merging steps
         image_combos_to_process = [
@@ -234,12 +234,12 @@ def run(message: dict, murfey_db: Session, demo: bool = False) -> bool:
                     f"{result.series_name!r} series",
                     exc_info=True,
                 )
-                return False
+                return {"success": False, "requeue": False}
         logger.info(
             "Successfully requested image alignment and merging job for "
             f"{result.series_name!r} series"
         )
-        return True
+        return {"success": True}
 
     finally:
         murfey_db.close()

src/murfey/workflows/notifications/__init__.py

@@ -8,7 +8,7 @@
 
 def notification_setup(
     message: dict, murfey_db: Session, num_instances_between_triggers: int = 500
-) -> bool:
+) -> dict[str, bool]:
     parameters: Dict[str, Tuple[float, float]] = {}
     for k in message.keys():
         parameter_name = ""
@@ -48,4 +48,4 @@ def notification_setup(
     murfey_db.add_all(existing_notification_parameters + new_notification_parameters)
     murfey_db.commit()
     murfey_db.close()
-    return True
+    return {"success": True}

src/murfey/workflows/spa/flush_spa_preprocess.py

@@ -306,15 +306,17 @@ def _flush_position_analysis(
     return register_foil_hole(session_id, gs.id, foil_hole_parameters, murfey_db)
 
 
-def flush_spa_preprocess(message: dict, murfey_db: Session, demo: bool = False) -> bool:
+def flush_spa_preprocess(
+    message: dict, murfey_db: Session, demo: bool = False
+) -> dict[str, bool]:
     session_id = message["session_id"]
     stashed_files = murfey_db.exec(
         select(PreprocessStash)
         .where(PreprocessStash.session_id == session_id)
         .where(PreprocessStash.tag == message["tag"])
     ).all()
     if not stashed_files:
-        return True
+        return {"success": True}
 
     murfey_session = murfey_db.exec(
         select(MurfeySession).where(MurfeySession.id == message["session_id"])
@@ -348,7 +350,7 @@ def flush_spa_preprocess(message: dict, murfey_db: Session, demo: bool = False)
         logger.warning(
             f"No SPA processing parameters found for client processing job ID {collected_ids[2].id}"
         )
-        return False
+        return {"success": False, "requeue": False}
 
     murfey_ids = _murfey_id(
         collected_ids[3].id,
@@ -444,4 +446,4 @@ def flush_spa_preprocess(message: dict, murfey_db: Session, demo: bool = False)
             )
     murfey_db.commit()
     murfey_db.close()
-    return True
+    return {"success": True}

src/murfey/workflows/spa/picking.py

@@ -440,7 +440,7 @@ def _check_notifications(message: dict, murfey_db: Session) -> None:
     return None
 
 
-def particles_picked(message: dict, murfey_db: Session) -> bool:
+def particles_picked(message: dict, murfey_db: Session) -> dict[str, bool]:
     movie = murfey_db.exec(
         select(Movie).where(Movie.murfey_id == message["motion_correction_id"])
     ).one()
@@ -460,4 +460,4 @@ def particles_picked(message: dict, murfey_db: Session) -> bool:
         processing_job=_pj_id(message["program_id"], murfey_db)
     ).inc()
     _check_notifications(message, murfey_db)
-    return True
+    return {"success": True}