diff --git a/d2d_development/README.md b/d2d_development/README.md index 7dc7360..160d6d8 100644 --- a/d2d_development/README.md +++ b/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': [], diff --git a/d2d_development/d2d_development/push.py b/d2d_development/d2d_development/push.py index 2c01a26..31b601f 100644 --- a/d2d_development/d2d_development/push.py +++ b/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,6 +291,7 @@ 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. @@ -298,6 +299,8 @@ def _push_data_points( ---------- 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. diff --git a/d2d_development/tests/test_push.py b/d2d_development/tests/test_push.py index d05bd3f..7d918c6 100644 --- a/d2d_development/tests/test_push.py +++ b/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())