Addressed the docstring and idds client function call ments from the PR review

Author: zhaoyu

doc/changes/DM-52866.feature.rst

@@ -1 +1 @@
-Update bps report function to aggregate panda task slices into task labels
+Update "bps report" function to aggregate panda task slices into task labels

python/lsst/ctrl/bps/panda/panda_service.py

@@ -55,6 +55,7 @@
     extract_taskname,
     get_idds_client,
     get_idds_result,
+    idds_call_with_check,
 )
 from lsst.resources import ResourcePath
 from lsst.utils.timer import time_this
@@ -174,12 +175,12 @@ def report(
             return run_reports, message
 
         idds_client = get_idds_client(self.config)
-        ret = idds_client.get_requests(request_id=wms_workflow_id, with_detail=True)
-        _LOG.debug("PanDA get workflow status returned = %s", str(ret))
-
-        request_status = ret[0]
-        if request_status != 0:
-            raise RuntimeError(f"Error to get workflow status: {ret} for id: {wms_workflow_id}")
+        ret = idds_call_with_check(
+            idds_client.get_requests,
+            func_name="get workflow status",
+            request_id=wms_workflow_id,
+            with_detail=True,
+        )
 
         tasks = ret[1][1]
         if not tasks:
@@ -268,8 +269,6 @@ def report(
             totaljobs = task.get("output_total_files", 0)
             wms_report.total_number_jobs += totaljobs
 
-            taskstatus = {}
-
             # --- If task failed/subfinished, fetch exit codes ---
             if status in ("SubFinished", "Failed") and not task_name.startswith("build_task"):
                 transform_workload_id = task.get("transform_workload_id")
@@ -279,18 +278,12 @@ def report(
                     nfailed = task.get("output_failed_files", 0)
                     exit_codes_all[tasklabel] = [1] * nfailed
                     if return_exit_codes:
-                        new_ret = idds_client.get_contents_output_ext(
-                            request_id=wms_workflow_id, workload_id=transform_workload_id
+                        new_ret = idds_call_with_check(
+                            idds_client.get_contents_output_ext,
+                            func_name=f"get task {transform_workload_id} detail",
+                            request_id=wms_workflow_id,
+                            workload_id=transform_workload_id,
                         )
-                        _LOG.debug(
-                            "PanDA get task %s detail returned = %s",
-                            transform_workload_id,
-                            str(new_ret),
-                        )
-                        if new_ret[0] != 0:
-                            raise RuntimeError(
-                                f"Error getting workflow status: {new_ret} for id: {wms_workflow_id}"
-                            )
                         # task_info is a dictionary of len 1 that contains
                         # a list of dicts containing panda job info
                         task_info = new_ret[1][1]
@@ -302,15 +295,16 @@ def report(
                                 if j.get("trans_exit_code") not in (None, 0, "0")
                             ]
                             if nfailed > 0 and len(exit_codes_all[tasklabel]) == 0:
-                                _LOG.warning(
+                                _LOG.debug(
                                     f"No exit codes in iDDS task info for workload {transform_workload_id}"
                                 )
                         else:
-                            err_msg = "Unexpected iDDS task info for workload "
-                            err_msg += f"{transform_workload_id}: {task_info}"
-                            raise RuntimeError(err_msg)
+                            raise RuntimeError(
+                                f"Unexpected iDDS task info for workload {transform_workload_id}: {task_info}"
+                            )
 
             # --- Aggregate job states ---
+            taskstatus = {}
             mapped_states = state_map.get(status, [])
             for state in WmsStates:
                 njobs = 0

python/lsst/ctrl/bps/panda/utils.py

@@ -35,6 +35,7 @@
     "extract_taskname",
     "get_idds_client",
     "get_idds_result",
+    "idds_call_with_check",
 ]
 
 import binascii
@@ -79,32 +80,34 @@
 
 
 def extract_taskname(s: str) -> str:
-    """
-    Extract the task name from a string that follows a pattern
+    """Extract the task name from a string that follows a pattern
     CampaignName_timestamp_TaskNumber_TaskLabel_ChunkNumber.
 
     Parameters
     ----------
-    s:  `str`
+    s : `str`
         The input string from which to extract the task name.
 
     Returns
     -------
-    `str`:
+    taskname : `str`
         The extracted task name as per the rules above.
     """
     # remove surrounding quotes/spaces if present
     s = s.strip().strip("'\"")
 
     # find all occurrences of underscore + digits + underscore,
     # take the last one
-    matches = list(re.finditer(r"_(\d+)_", s))
+    matches = re.findall(r"_(\d+)_", s)
     if matches:
-        last = matches[-1]
-        return s[last.end() :]
+        last_number = matches[-1]
+        last_pos = s.rfind(f"_{last_number}_") + len(f"_{last_number}_")
+        taskname = s[last_pos:]
+        return taskname
 
     # fallback: if no such pattern, return everything
-    return s
+    taskname = s
+    return taskname
 
 
 def aggregate_by_basename(job_summary, exit_code_summary, run_summary):
@@ -113,35 +116,31 @@ def aggregate_by_basename(job_summary, exit_code_summary, run_summary):
 
     Parameters
     ----------
-    job_summary : `dict`[`str`, `dict`[`str`, `int`]]
+    job_summary : `dict` [`str`, `dict` [`str`, `int`]]
         A mapping of job labels to state-count mappings.
-    exit_code_summary : `dict`[`str`, `list`[`int`]]
+    exit_code_summary : `dict` [`str`, `list` [`int`]]
         A mapping of job labels to lists of exit codes.
     run_summary : `str`
         A semicolon-separated string of job summaries
         where each entry has the format "<label>:<count>".
 
     Returns
     -------
-    `tuple`[`dict`[`str`, `dict`[`str`, `int`]],
-            `dict`[`str`, `list`[`int`]], `str`]
-        aggregated_jobs : `dict`
-            A dictionary mapping each base label to the summed
-            job state counts across all matching labels.
-        aggregated_exits : `dict`
-            A dictionary mapping each base label to a combined list of
-            exit codes from all matching labels.
-        aggregated_run: `str`
-            A semicolon-separated string with aggregated job counts
-            by base label.
+    aggregated_jobs : `dict` [`str`, `dict` [`str`, `int`]]
+        A dictionary mapping each base label to the summed job state counts
+        across all matching labels.
+    aggregated_exits : `dict` [`str`, `list` [`int`]]
+        A dictionary mapping each base label to a combined list of exit codes
+        from all matching labels.
+    aggregated_run : `str`
+        A semicolon-separated string with aggregated job counts by base label.
     """
 
     def base_label(label):
         return re.sub(r"_\d+$", "", label)
 
     aggregated_jobs = {}
     aggregated_exits = {}
-    aggregated_run = ""
 
     for label, states in job_summary.items():
         base = base_label(label)
@@ -155,7 +154,6 @@ def base_label(label):
         aggregated_exits.setdefault(base, []).extend(codes)
 
     aggregated = {}
-    ordered_labels = []
     for entry in run_summary.split(";"):
         entry = entry.strip()
         if not entry:
@@ -167,12 +165,9 @@ def base_label(label):
             continue
 
         base = base_label(label)
-        if base not in aggregated:
-            aggregated[base] = 0
-            ordered_labels.append(base)
-        aggregated[base] += num
+        aggregated[base] = aggregated.get(base, 0) + num
 
-    aggregated_run = ";".join(f"{base}:{aggregated[base]}" for base in ordered_labels)
+    aggregated_run = ";".join(f"{base}:{count}" for base, count in aggregated.items())
     return aggregated_jobs, aggregated_exits, aggregated_run
 
 
@@ -294,6 +289,40 @@ def get_idds_result(ret):
     return status, result, error
 
 
+def idds_call_with_check(func, *, func_name: str, request_id: int, **kwargs):
+    """Call an iDDS client function, log, and check the return code.
+
+    Parameters
+    ----------
+    func : callable
+        The iDDS client function to call.
+    func_name : `str`
+        Name used for logging.
+    request_id : `int`
+        The request or workflow ID.
+    **kwargs
+        Additional keyword arguments passed to the function.
+
+    Returns
+    -------
+    ret : `Any`
+        The return value from the iDDS client function.
+    """
+    call_kwargs = dict(kwargs)
+    if request_id is not None:
+        call_kwargs["request_id"] = request_id
+
+    ret = func(**call_kwargs)
+
+    _LOG.debug("PanDA %s returned = %s", func_name, str(ret))
+
+    request_status = ret[0]
+    if request_status != 0:
+        raise RuntimeError(f"Error calling {func_name}: {ret} for id: {request_id}")
+
+    return ret
+
+
 def _make_pseudo_filename(config, gwjob):
     """Make the job pseudo filename.