inference + alert clearing for reprocess_run

dataquality/clients/api.py

@@ -619,20 +619,51 @@ def get_column_distribution(
         body = {"task": task, "filter_params": filter_params or {}}
         return self.make_request(RequestType.POST, url, body=body, params=params)
 
-    def get_xray_cards(
+    def get_alerts(
         self,
         project_name: str,
         run_name: str,
         split: str,
         inference_name: Optional[str] = None,
     ) -> List[Dict[str, str]]:
-        """Queries API for xray cards for a run/split"""
+        """Queries API for alerts for a run/split"""
         project, run = self._get_project_run_id(project_name, run_name)
         path = Route.content_path(project, run, split)
-        url = f"{config.api_url}/{path}/{Route.xray}"
+        url = f"{config.api_url}/{path}/{Route.alerts}"
         params = {"inference_name": inference_name} if inference_name else None
         return self.make_request(RequestType.GET, url, params=params)
 
+    def delete_alerts_for_split(
+        self, project_id: UUID4, run_id: UUID4, split: str
+    ) -> None:
+        path = Route.content_path(project_id, run_id, split)
+        url = f"{config.api_url}/{path}/{Route.alerts}"
+        alerts = []
+        if split == "inference":
+            inference_names = self.get_inference_names(project_id, run_id)
+            for inf_name in inference_names["inference_names"]:
+                params = {"inference_name": inf_name}
+                res = self.make_request(RequestType.GET, url, params=params)
+                alerts.extend([alert["id"] for alert in res])
+        else:
+            res = self.make_request(RequestType.GET, url)
+            alerts.extend([alert["id"] for alert in res])
+        path = Route.content_path(project_id, run_id, split)
+        for alert_id in alerts:
+            url = f"{config.api_url}/{path}/{Route.alerts}/{alert_id}"
+            self.make_request(RequestType.DELETE, url)
+
+    def delete_alerts(
+        self,
+        project_name: str,
+        run_name: str,
+    ) -> None:
+        """Delete all alerts for a run"""
+        project_id, run_id = self._get_project_run_id(project_name, run_name)
+        for split in self.get_splits(project_id, run_id)["splits"]:
+            self.delete_alerts_for_split(project_id, run_id, split)
+        print(f"All alerts removed for run {project_name}/{run_name}")
+
     def get_edits(
         self,
         project_name: str,

dataquality/internal.py

@@ -10,7 +10,7 @@
 
 
 def reprocess_run(
-    project_name: str, run_name: str, alerts: bool = False, wait: bool = True
+    project_name: str, run_name: str, alerts: bool = True, wait: bool = True
 ) -> None:
     """Reprocesses a run that has already been processed by Galileo
 
@@ -19,9 +19,8 @@ def reprocess_run(
 
     :param project_name: The name of the project
     :param run_name: The name of the run
-    :param alerts: Whether to create the alerts. Currently, this will not delete the
-        existing alerts, so they will be duplicated if they already exist. This
-        feature will come soon
+    :param alerts: Whether to create the alerts. If True, all alerts for the run will
+        be removed, and recreated during processing. Default True
     :param wait: Whether to wait for the run to complete processing on the server. If
         True, this will block execution, printing out the status updates of the run.
         Useful if you want to know exactly when your run completes. Otherwise, this will
@@ -52,8 +51,10 @@ def reprocess_run(
         tasks = api_client.get_tasks_for_run(project_name, run_name)
     if task_type == TaskType.text_ner:
         # In NER, dq.metrics.get_labels_for_run will return the _full_ label set in NER
-        # form (ie B-PER, I-PER, O-PER, B-LOC, etc) which is needed for processeing
+        # form (ie B-PER, I-PER, O-PER, B-LOC, etc) which is needed for processing
         ner_labels = dataquality.metrics.get_labels_for_run(project_name, run_name)
+    if alerts:
+        api_client.delete_alerts(project_name, run_name)
     body = dict(
         project_id=str(project),
         run_id=str(run),
@@ -62,6 +63,13 @@ def reprocess_run(
         task_type=task_type,
         ner_labels=ner_labels,
         xray=alerts,
+        # We set the job name to inference and non_inference_logged to True because
+        # This will force the server to first reprocess the non-inference splits,
+        # and then reprocess all of the inference splits. If there are no inference
+        # splits, this will still work as expected, inference will just be skipped
+        job_name="inference",
+        process_existing_inference_runs=True,
+        non_inference_logged=True,
     )
     res = api_client.make_request(
         RequestType.POST, url=f"{config.api_url}/{Route.jobs}", body=body

dataquality/metrics.py

@@ -643,15 +643,15 @@ def get_raw_data(
     )
 
 
-def get_xray_cards(
+def get_alerts(
     project_name: str, run_name: str, split: Split, inference_name: Optional[str] = None
 ) -> List[Dict[str, str]]:
-    """Get xray cards for a project/run/split
+    """Get alerts for a project/run/split
 
-    Xray cards are automatic insights calculated and provided by Galileo on your data
+    Alerts are automatic insights calculated and provided by Galileo on your data
     """
     split = conform_split(split)
-    return api_client.get_xray_cards(project_name, run_name, split, inference_name)
+    return api_client.get_alerts(project_name, run_name, split, inference_name)
 
 
 def get_labels_for_run(

dataquality/schemas/route.py

@@ -33,7 +33,7 @@ class Route(str, Enum):
     groupby = "insights/groupby"
     metrics = "metrics"
     distribution = "insights/distribution"
-    xray = "insights/xray"
+    alerts = "insights/alerts"
     export = "export"
     edits = "edits"
     export_edits = "edits/export"