fix(push): deleted data points (NA) logging

d2d_development/README.md

@@ -188,7 +188,8 @@ pusher.push_data(df)
 
 After calling `push_data`, the `DHIS2Pusher` instance provides detailed results of the push operation in its `summary` attribute. This dictionary contains:
 
-- `import_counts`: Number of data points imported, updated, ignored, or deleted (dict with keys: `imported`, `updated`, `ignored`, `deleted`).
+- `import_counts`: Number of data points imported, updated, ignored, or deleted from the valid-data push (dict with keys: `imported`, `updated`, `ignored`, `deleted`).
+- `import_counts_delete`: Same shape as `import_counts`, but for the separate push that sets NA-valued data points to be deleted.
 - `import_options`: The options used for the import (strategy, dry run, etc).
 - `import_errors`: List of errors, conflicts, or error reports returned by DHIS2 or encountered during the push.
 - `rejected_datapoints`: List of data points that were rejected by DHIS2 itself (e.g. a conflict on a specific value), identified from the API response.
@@ -202,6 +203,7 @@ print(pusher.summary)
 # Example output:
 # {
 #   'import_counts': {'imported': 1, 'updated': 0, 'ignored': 0, 'deleted': 0},
+#   'import_counts_delete': {'imported': 0, 'updated': 0, 'ignored': 0, 'deleted': 1},
 #   'import_options': {'importStrategy': 'CREATE_AND_UPDATE', 'dryRun': False, ...},
 #   'import_errors': [],
 #   'rejected_datapoints': [],

d2d_development/d2d_development/push.py

@@ -149,7 +149,6 @@ def _push_valid(self, data_points_valid: pl.DataFrame) -> None:
         periods = sorted(data_points_valid["period"].unique().to_list())
         self._log_message(f"Period(s): {', '.join(periods)}.")
         self._push_data_points(data_point_list=self._serialize_data_points(data_points_valid))
-        self._log_message(f"Data points push summary:  {self.summary['import_counts']}")
 
     def _push_to_delete(self, data_points_to_delete: pl.DataFrame) -> None:
         """Push data points with NA values to DHIS2 to delete them."""
@@ -158,8 +157,9 @@ def _push_to_delete(self, data_points_to_delete: pl.DataFrame) -> None:
 
         self._log_message(f"Pushing {len(data_points_to_delete)} data points with NA values.")
         self._log_ignored_or_na(data_points_to_delete, is_na=True)
-        self._push_data_points(data_point_list=self._serialize_data_points(data_points_to_delete))
-        self._log_message(f"Data points delete summary: {self.summary['import_counts']}")
+        self._push_data_points(
+            data_point_list=self._serialize_data_points(data_points_to_delete), summary_key="import_counts_delete"
+        )
 
     def _initialize_cache(self, cache_path: Path | None) -> None:
         """Initialize the cache for tracking pushed data points."""
@@ -291,13 +291,16 @@ def _post(self, chunk: list[dict]) -> requests.Response:
     def _push_data_points(
         self,
         data_point_list: list[dict],
+        summary_key: str = "import_counts",
     ) -> None:
         """Push data points to DHIS2 in chunks, handling responses and logging progress.
 
         Parameters
         ----------
         data_point_list: list[dict]
             A list of dictionaries, each representing a data point formatted for DHIS2.
+        summary_key: str
+            Key of `self.summary` to accumulate the import counts into. Defaults to "import_counts".
         """
         total_data_points = len(data_point_list)
         processed_points = 0
@@ -312,13 +315,13 @@ def _push_data_points(
                 response = self._safe_json(r)
 
                 if response:
-                    self._update_import_counts(response)
+                    self._update_import_counts(response, summary_key=summary_key)
 
             except requests.exceptions.RequestException as e:
                 self._raise_server_errors(r)  # Stop the process if there's a server error
                 response = self._safe_json(r)
                 if response:
-                    self._update_import_counts(response)
+                    self._update_import_counts(response, summary_key=summary_key)
                 else:
                     # No response JSON, at least log the request error msg
                     self.summary["import_errors"].extend(
@@ -335,15 +338,13 @@ def _push_data_points(
                 progress_pct = (processed_points / total_data_points) * 100
                 self._log_message(
                     f"{processed_points} / {total_data_points} data points ({progress_pct:.1f}%) "
-                    f" summary: {self.summary['import_counts']}"
+                    f" summary: {self.summary[summary_key]}"
                 )
                 last_logged_at = processed_points
 
         # Final summary
-        self._log_message(
-            f"{processed_points} / {total_data_points} data points processed."
-            f" Final summary: {self.summary['import_counts']}"
-        )
+        self._log_message(f"{processed_points} / {total_data_points} data points processed.")
+        self._log_message(f"Push summary: {self.summary[summary_key]}")
 
     def _raise_server_errors(self, r: requests.Response) -> None:
         """Check if the response indicates a server error (stop process)."""
@@ -365,6 +366,7 @@ def _reset_summary(self) -> None:
         """Reset the summary dictionary to its initial state before starting a new push operation."""
         self.summary = {
             "import_counts": {"imported": 0, "updated": 0, "ignored": 0, "deleted": 0},
+            "import_counts_delete": {"imported": 0, "updated": 0, "ignored": 0, "deleted": 0},
             "import_options": {},
             "import_errors": [],
             "rejected_datapoints": [],
@@ -395,7 +397,7 @@ def _safe_json(self, r: requests.Response) -> dict | None:
         except (ValueError, json.JSONDecodeError):
             return None
 
-    def _update_import_counts(self, response: dict) -> None:
+    def _update_import_counts(self, response: dict, summary_key: str = "import_counts") -> None:
         """Update the import counts in the summary dictionary based on the response from DHIS2."""
         if not response:
             return
@@ -406,7 +408,7 @@ def _update_import_counts(self, response: dict) -> None:
         else:
             import_counts = {}
         for key in ["imported", "updated", "ignored", "deleted"]:
-            self.summary["import_counts"][key] += import_counts.get(key, 0)
+            self.summary[summary_key][key] += import_counts.get(key, 0)
 
     def _extract_conflicts(self, response: dict, chunk: list) -> None:
         """Extract all conflicts and errorReports from a DHIS2 API response.

d2d_development/tests/test_push.py

@@ -195,6 +195,46 @@ def test_push_data_point():
         }
 
 
+def test_push_data_import_counts_separated_from_delete_counts():
+    """Test that valid-push and delete-push counts are tracked in separate summary buckets."""
+    data_points = DHIS2Extractor(dhis2_client=MockDHIS2Client()).data_elements._retrieve_data(
+        data_elements=["AAA111", "DELETE1"], org_units=[], period="202501"
+    )
+
+    pusher = DHIS2Pusher(dhis2_client=MockDHIS2Client())
+
+    mock_delete_response = {
+        "httpStatus": "OK",
+        "httpStatusCode": 200,
+        "status": "OK",
+        "message": "Import was successful.",
+        "response": {
+            "responseType": "ImportSummary",
+            "status": "SUCCESS",
+            "importCount": {"imported": 0, "updated": 0, "ignored": 0, "deleted": 1},
+            "conflicts": [],
+            "rejectedIndexes": [],
+        },
+    }
+
+    # First POST call is the valid-points push, second is the delete push
+    with patch.object(
+        pusher.dhis2_client.api.session,
+        "post",
+        side_effect=[
+            MockDHIS2Response(MOCK_DHIS2_OK_RESPONSE),
+            MockDHIS2Response(mock_delete_response),
+        ],
+    ):
+        pusher.push_data(data_points)
+
+    assert pusher.summary["import_counts"]["imported"] == 1
+    assert pusher.summary["import_counts"]["deleted"] == 0
+
+    assert pusher.summary["import_counts_delete"]["imported"] == 0
+    assert pusher.summary["import_counts_delete"]["deleted"] == 1
+
+
 def test_push_data_points_connection_error():
     """Test the error handling of error 503 to DHIS2."""
     pusher = DHIS2Pusher(dhis2_client=MockDHIS2Client())