added logging and error handling

Author: Sandip117

dypxFlow.py

@@ -29,7 +29,7 @@
 logger.remove()
 logger.add(sys.stderr, format=logger_format)
 
-__version__ = '1.0.6'
+__version__ = '1.0.7'
 
 DISPLAY_TITLE = r"""
        _           _                ______ _               
@@ -166,6 +166,7 @@ def main(options: Namespace, inputdir: Path, outputdir: Path):
         df_clean = df.fillna('')
         l_job = create_query(df_clean)
         d_df = []
+        pipeline_errors = False
         if int(options.thread):
             with concurrent.futures.ThreadPoolExecutor(max_workers=int(options.maxThreads)) as executor:
                 results: Iterator = executor.map(lambda t: register_and_anonymize(options, t, options.wait), l_job)
@@ -174,15 +175,20 @@ def main(options: Namespace, inputdir: Path, outputdir: Path):
             # executor.shutdown(wait=True)
         else:
             for d_job in l_job:
-                status = register_and_anonymize(options, d_job)
+                response = register_and_anonymize(options, d_job)
                 row = d_job["raw"]
                 row.update(d_job["push"])
-                row["status"] = status
+                row["status"] = response['status']
                 d_df.append(row)
+                if response.get('error'):
+                    pipeline_errors = True
 
         csv_file = os.path.join(options.outputdir, input_file.name)
         df = pd.DataFrame(d_df)
         df.to_csv(csv_file, index=False)
+        if pipeline_errors:
+            LOG(f"ERROR while running pipelines.")
+            sys.exit(1)
 
 
 if __name__ == '__main__':
@@ -192,6 +198,7 @@ def register_and_anonymize(options: Namespace, d_job: dict, wait: bool = False):
     1) Search through PACS for series and register in CUBE
     2) Run anonymize and push workflow on the registered series
     """
+    resp = {}
     d_job["pull"] = {
         "url": options.PACSurl,
         "pacs": options.PACSname
@@ -202,7 +209,9 @@ def register_and_anonymize(options: Namespace, d_job: dict, wait: bool = False):
         d_ret = cube_con.anonymize(d_job, options.pluginInstanceID)
     else:
         d_ret = d_job["push"]
-    return d_ret['status']
+
+
+    return d_ret
 
 
 def health_check(options) -> bool:

pipeline.py

@@ -76,9 +76,19 @@ def __init__(self, client):
         stop=stop_after_attempt(5),
         reraise=True
     )
-    def make_request(self, method, endpoint, **kwargs):
+    def make_request(self, method: str, endpoint: str, **kwargs):
         url = f"{self.api_base}{endpoint}"
-        response = requests.request(method, url, headers=self.headers, auth=self.auth, timeout=5, **kwargs)
+        response = requests.request(method, url, headers=self.headers, auth=self.auth, timeout=30, **kwargs)
+        response.raise_for_status()
+
+        try:
+            return response.json().get("collection", {}).get("items", [])
+        except ValueError:
+            return response.text
+
+    def post_request(self, endpoint: str, **kwargs):
+        url = f"{self.api_base}{endpoint}"
+        response = requests.request("POST", url, headers=self.headers, auth=self.auth, timeout=30, **kwargs)
         response.raise_for_status()
 
         try:
@@ -93,6 +103,7 @@ def get_pipeline_id(self, name: str) -> int:
         """Fetch pipeline ID by name."""
         logger.info(f"Fetching ID for pipeline: {name}")
         response = self.make_request("GET", f"/pipelines/search/?name={name}")
+
         for item in response:
             for field in item.get("data", []):
                 if field.get("name") == "id":
@@ -113,7 +124,7 @@ def post_workflow(self, pipeline_id: int, previous_id: int, params: list[dict])
             "previous_plugin_inst_id": previous_id,
             "nodes_info": json.dumps(params)
         }
-        return self.make_request("POST", f"/pipelines/{pipeline_id}/workflows/", json=payload)
+        return self.post_request( f"/pipelines/{pipeline_id}/workflows/", json=payload)
 
     def run_pipeline(self, pipeline_name: str, previous_inst: int, pipeline_params: dict):
         """
@@ -129,7 +140,7 @@ def run_pipeline(self, pipeline_name: str, previous_inst: int, pipeline_params:
             nodes_info = compute_workflow_nodes_info(default_params, include_all_defaults=True)
             updated_params = update_plugin_parameters(nodes_info, pipeline_params)
             workflow = self.post_workflow(pipeline_id=pipeline_id, previous_id=previous_inst, params=updated_params)
-            logger.info(f"Workflow posted successfully: {workflow}")
+            logger.info(f"Workflow posted successfully")
             return {"status": "Pipeline running"}
         except Exception as ex:
             logger.error(f"Running pipeline failed due to: {ex}")