Push cache

.github/workflows/ci.yaml

@@ -13,9 +13,11 @@ jobs:
       - uses: actions/checkout@v4
       - uses: astral-sh/ruff-action@v3
         with:
+          version: "0.15.22"
           args: check
       - uses: astral-sh/ruff-action@v3
         with:
+          version: "0.15.22"
           args: format --check
 
   test:

.pre-commit-config.yaml

@@ -10,7 +10,7 @@ repos:
     -   id: check-added-large-files
 
 -   repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.15.1
+    rev: v0.15.22
     hooks:
       - id: ruff-check
         args: [ --fix ]

d2d_development/README.md

@@ -154,6 +154,7 @@ When initializing `DHIS2Pusher`, you can configure the following parameters:
 - `max_post`: Maximum number of data points per POST request (default: `500`).
 - `logging_interval`: Log progress every N data points (default: `50000`).
 - `logger`: Optional custom logger instance.
+- `cache_path`: Optional `Path`. If provided, enables push caching so datapoints that haven't changed since the last successful push are skipped (default: `None`, caching disabled). See [Caching Pushed Datapoints](#caching-pushed-datapoints-avoid-re-pushing-unchanged-data) below.
 
 **Usage Example:**
 ```python
@@ -190,6 +191,7 @@ After calling `push_data`, the `DHIS2Pusher` instance provides detailed results
 - `import_counts`: Number of data points imported, updated, ignored, or deleted (dict with keys: `imported`, `updated`, `ignored`, `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.
 - `ignored_data_points`: List of data points that were ignored due to missing or invalid fields.
 - `delete_data_points`: List of data points that were marked for deletion (value is NA/null).
 
@@ -202,6 +204,7 @@ print(pusher.summary)
 #   'import_counts': {'imported': 1, 'updated': 0, 'ignored': 0, 'deleted': 0},
 #   'import_options': {'importStrategy': 'CREATE_AND_UPDATE', 'dryRun': False, ...},
 #   'import_errors': [],
+#   'rejected_datapoints': [],
 #   'ignored_data_points': [],
 #   'delete_data_points': []
 # }
@@ -216,4 +219,47 @@ if pusher.summary["ignored_data_points"]:
         print(dp)
 ```
 
----
+---
+
+### Caching Pushed Datapoints (avoid re-pushing unchanged data)
+
+**Description:**
+`DHIS2Pusher` can optionally track which datapoints have already been successfully pushed to DHIS2, so that subsequent runs only push datapoints that are new or have changed. This is useful for pipelines that run repeatedly (e.g. on a schedule) over largely the same data, where re-sending unchanged values on every run wastes time and DHIS2 API calls.
+
+A datapoint is identified by its key columns (`dx`, `period`, `orgUnit`, `categoryOptionCombo`, `attributeOptionCombo`). It's considered unchanged, and therefore skipped, only if its `value` is identical to the last successfully pushed value for that same key. Any difference in `value` - including a change to or from a null/NA value (a deletion) - is treated as new and gets pushed.
+
+**Enabling the cache:**
+Pass a `cache_path` when creating `DHIS2Pusher`:
+
+```python
+from pathlib import Path
+from d2d_development.push import DHIS2Pusher
+
+pusher = DHIS2Pusher(
+	dhis2_client,
+	cache_path=Path("/path/to/persistent/cache"),  # must persist across runs
+	dry_run=False,  # the cache is only used when dry_run is False
+)
+
+pusher.push_data(df)
+```
+
+If `cache_path` is omitted (the default), caching is disabled entirely and `push_data` behaves exactly as without this feature - every call pushes the full input.
+
+**How it works:**
+- Before pushing, `push_data` filters out any datapoint whose key and value exactly match what's cached from a previous run.
+- After a push, only datapoints that DHIS2 actually accepted are recorded in the cache - a datapoint rejected by DHIS2 (see `rejected_datapoints` in the summary above) is left out, so it will be attempted again on the next run instead of being silently treated as done.
+- The cache is stored on disk as one `cache_<period>.parquet` file per period inside `cache_path`, so `cache_path` should point to a location that persists between pipeline runs (not a temporary directory), otherwise the cache provides no benefit.
+- If `dry_run` is `True` (the default), the cache is not used at all, even when `cache_path` is set: datapoints are not filtered against it, and it is not updated with the datapoints from this run. This keeps a dry run from marking datapoints as pushed when nothing was actually sent to DHIS2. Pass `dry_run=False` to enable the cache.
+
+**Performance tip:** if you push several distinct datasets (e.g. different extract groups) to the same DHIS2 instance, use a separate `cache_path` per dataset rather than sharing one. Each `cache_<period>.parquet` file is fully read and rewritten on every push, so a shared cache grows with the combined size of every dataset using it, not just the size of the one being pushed.
+
+```python
+base_cache_dir = Path("/path/to/persistent/cache")
+
+pusher_dataset_a = DHIS2Pusher(dhis2_client, cache_path=base_cache_dir / "dataset_a", dry_run=False)
+pusher_dataset_a.push_data(df_a)
+
+pusher_dataset_b = DHIS2Pusher(dhis2_client, cache_path=base_cache_dir / "dataset_b", dry_run=False)
+pusher_dataset_b.push_data(df_b)
+```

d2d_development/d2d_development/exceptions.py

@@ -4,3 +4,7 @@ class ExtractorError(Exception):
 
 class PusherError(Exception):
     """Custom exception for all DHIS2Pusher errors."""
+
+
+class PusherCacheError(Exception):
+    """Custom exception for all DHIS2Pusher cache errors."""

d2d_development/d2d_development/push.py

@@ -1,5 +1,6 @@
 import json
 import logging
+from pathlib import Path
 
 import pandas as pd
 import polars as pl
@@ -8,6 +9,7 @@
 
 from .data_models import DataPointModel
 from .exceptions import PusherError
+from .push_cache import DHIS2PushCache
 from .utils import log_message
 
 
@@ -22,6 +24,7 @@ def __init__(
         max_post: int = 500,
         logging_interval: int = 50000,
         logger: logging.Logger | None = None,
+        cache_path: Path | None = None,
     ):
         """Initialize the DHIS2Pusher."""
         self.dhis2_client = dhis2_client
@@ -38,6 +41,7 @@ def __init__(
         self._reset_summary()
         self.logger = logger if logger else logging.getLogger(__name__)
         self.log_function = log_message
+        self._initialize_cache(cache_path)
 
     def push_data(
         self,
@@ -68,12 +72,16 @@ def push_data(
             self._log_message("Input DataFrame is empty. No data to push.")
             return
 
+        if self.push_cache:
+            df_data = self.push_cache.filter_new(df_data)
+
         valid, to_delete, to_ignore = self._classify_data_points(df_data)
 
         self._push_valid(valid)
         self._push_to_delete(to_delete)
         self._log_summary_errors()
         self._log_ignored_or_na(to_ignore)
+        self._update_cache_with(pl.concat([valid, to_delete]))
 
     def _validate_input_data(self, df_data: pl.DataFrame) -> None:
         """Validate that the input DataFrame contains all mandatory fields.
@@ -134,7 +142,9 @@ def _push_valid(self, data_points_valid: pl.DataFrame) -> None:
             self._log_message("No data to push.")
             return
 
-        self._log_message(f"Pushing {len(data_points_valid)} data points.")
+        self._log_message(f"Pushing {data_points_valid.height} data points.")
+        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']}")
 
@@ -148,6 +158,49 @@ def _push_to_delete(self, data_points_to_delete: pl.DataFrame) -> None:
         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']}")
 
+    def _initialize_cache(self, cache_path: Path | None) -> None:
+        """Initialize the cache for tracking pushed data points."""
+        if cache_path is None or self.dry_run:
+            self.push_cache = None
+            return
+
+        self.push_cache = DHIS2PushCache(cache_path=cache_path, logger=self.logger)
+
+    def _update_cache_with(self, data: pl.DataFrame) -> None:
+        """Update the cache with the latest pushed data points."""
+        if self.push_cache is not None:
+            self.push_cache.mark_pushed(self._remove_rejected_points(data))
+
+    def _remove_rejected_points(self, data: pl.DataFrame) -> pl.DataFrame:
+        """Remove data points that were rejected by DHIS2 from the DataFrame before updating the cache.
+
+        `rejected_datapoints` entries are the *serialized* DHIS2 payload (DHIS2-cased keys), so
+        rejected points are matched back to `data` by remapping to the internal key columns.
+
+        Returns
+        -------
+            pl.DataFrame: A DataFrame containing only the data points that were successfully pushed to DHIS2.
+        """
+        rejected = self.summary.get("rejected_datapoints", [])
+        if not rejected:
+            return data
+
+        key_columns = [c for c in self.mandatory_fields if c != "value"]
+        rejected_keys = pl.DataFrame(
+            [
+                {
+                    "dx": r["dataElement"],
+                    "period": r["period"],
+                    "org_unit": r["orgUnit"],
+                    "category_option_combo": r["categoryOptionCombo"],
+                    "attribute_option_combo": r["attributeOptionCombo"],
+                }
+                for r in rejected
+            ]
+        ).unique()
+
+        return data.join(rejected_keys, on=key_columns, how="anti")
+
     def _log_ignored_or_na(self, data_points: pl.DataFrame, is_na: bool = False):
         """Logs ignored or NA data points.
 
@@ -384,8 +437,6 @@ def _extract_conflicts(self, response: dict, chunk: list) -> None:
 
         # Extract rejected datapoints
         if rejected_indexes:
-            rejected_datapoints = [
-                {"index": idx, "datapoint": chunk[idx]} for idx in rejected_indexes if 0 <= idx < len(chunk)
-            ]
+            rejected_datapoints = [chunk[idx] for idx in rejected_indexes if 0 <= idx < len(chunk)]
             if rejected_datapoints:
                 self.summary.setdefault("rejected_datapoints", []).extend(rejected_datapoints)

d2d_development/d2d_development/push_cache.py

@@ -0,0 +1,171 @@
+import logging
+from pathlib import Path
+
+import polars as pl
+
+from .exceptions import PusherCacheError
+from .utils import log_message
+
+
+class DHIS2PushCache:
+    """Tracks datapoints already pushed to DHIS2 to avoid re-pushing unchanged records.
+
+    A datapoint is identified by hashing the metadata columns of each row in the data to push.
+    Columns:
+     ["dx", "period", "org_unit", "category_option_combo", "attribute_option_combo"]
+    A row is only considered "pushed" when the value in the "value" column, is identical to the cached value.
+    A row is considered "new" if any the value columns has changed since the last push.
+    Any difference in the value column, including None, is treated as a new.
+    The cache data is updated with the new value after a successful push (mark_pushed(data pushed)).
+    """
+
+    def __init__(self, cache_path: Path, logger: logging.Logger | None = None, verbose: bool = False):
+        self.cache_path = Path(cache_path)
+        self.logger = logger if logger else logging.getLogger(__name__)
+        self.verbose = verbose
+        self._cache_data: pl.DataFrame | None = None
+        self._mandatory_fields = [
+            "dx",
+            "period",
+            "org_unit",
+            "category_option_combo",
+            "attribute_option_combo",
+            "value",
+        ]
+        self._log_function = log_message
+
+    def filter_new(self, data: pl.DataFrame) -> pl.DataFrame:
+        """Returns the rows of `data` that have not already been pushed.
+
+        Args:
+            data: Datapoints (serialized DataPointModel) about to be pushed.
+
+        Returns:
+            Subset of `data` whose rows have no exact match in the cache.
+        """
+        if data.height == 0:
+            self._log_message("No data to push.", log_current_run=self.verbose)
+            return data
+
+        self._load(self._resolve_data_periods(data))
+        if self._cache_data is None or self._cache_data.height == 0:
+            self._log_message("No cache data found. All rows will be pushed.", log_current_run=self.verbose)
+            return data
+
+        to_push = (
+            data.join(
+                self._cache_data.with_columns(pl.lit(value=True).alias("_cache_hit")),
+                on=[c for c in self._mandatory_fields if c != "value"],
+                how="left",
+                suffix="_cached",
+            )
+            .filter(pl.col("_cache_hit").is_null() | pl.col("value").ne_missing(pl.col("value_cached")))
+            .select(data.columns)
+        )
+
+        self._log_message(
+            f"Filtered {data.height - to_push.height} rows already pushed to DHIS2.", log_current_run=self.verbose
+        )
+        return to_push
+
+    def mark_pushed(self, data: pl.DataFrame) -> None:
+        """Records `data` as pushed, so filter_new() skips it on the next run.
+
+        Call this only after a push has been confirmed successful - marking
+        rows before that risks silently dropping datapoints whose push failed.
+
+        Args:
+            data: Datapoints that were just pushed successfully.
+        """
+        if data.height > 0:
+            self.cache_path.mkdir(parents=True, exist_ok=True)
+            data = data.select(self._mandatory_fields)
+            if self._cache_data is None:
+                self._create_cache(data)
+            else:
+                self._update_cache(data)
+
+    def _resolve_data_periods(self, data: pl.DataFrame) -> list[str]:
+        """Returns the periods present in the data.
+
+        Args:
+            data: Datapoints to be pushed.
+
+        Returns:
+            The periods present in the data.
+        """
+        return data["period"].unique().sort().to_list()
+
+    def _load(self, periods: list[str]) -> None:
+        """Reads cache_<period>.parquet files matching `periods` into self._cache_data.
+
+        This always reads fresh from disk on every call - it does not skip periods
+        that were already loaded by a previous call.
+        """
+        if self.cache_path.exists():
+            file_paths = []
+            data_list = []
+            for period in periods:
+                file_paths.extend(self.cache_path.glob(f"cache_{period}.parquet"))
+            for fname in file_paths:
+                try:
+                    data_list.append(pl.read_parquet(fname))
+                except Exception as e:
+                    self._log_message(
+                        f"Failed to load cache file {fname}.",
+                        level="error",
+                        error_details=str(e),
+                    )
+            self._cache_data = pl.concat(data_list) if data_list else None
+        else:
+            self._log_message(
+                f"Cache path {self.cache_path} does not exist. No cache data loaded.", log_current_run=self.verbose
+            )
+
+    def _create_cache(self, data: pl.DataFrame) -> None:
+        """Creates a new cache from the provided data."""
+        periods = self._resolve_data_periods(data)
+        key_columns = [c for c in self._mandatory_fields if c != "value"]
+        for period in periods:
+            period_data = data.filter(pl.col("period") == period).unique(subset=key_columns, keep="last")
+            period_data.write_parquet(self.cache_path / f"cache_{period}.parquet")
+
+    def _update_cache(self, data_pushed: pl.DataFrame) -> None:
+        """Updates the cache with the newly pushed data.
+
+        For each period, cached rows whose datapoint key matches a newly pushed row
+        are dropped and replaced by the pushed row's new value; rows whose key was
+        not previously cached are simply added.
+        """
+        key_columns = [c for c in self._mandatory_fields if c != "value"]
+        periods = self._resolve_data_periods(data_pushed)
+        for period in periods:
+            period_data = data_pushed.filter(pl.col("period") == period).unique(subset=key_columns, keep="last")
+            cached_period_data = self._cache_data.filter(pl.col("period") == period)
+            other_periods_data = self._cache_data.filter(pl.col("period") != period)
+
+            # remove any cached rows matching the newly pushed rows, and then add the new ones.
+            updated_period_data = pl.concat(
+                [cached_period_data.join(period_data.select(key_columns), on=key_columns, how="anti"), period_data]
+            )
+            self._cache_data = pl.concat([other_periods_data, updated_period_data])
+
+            try:
+                updated_period_data.write_parquet(self.cache_path / f"cache_{period}.parquet")
+            except Exception as e:
+                self._log_message(
+                    f"Failed to update cache file for period {period}.",
+                    level="error",
+                    error_details=str(e),
+                )
+
+    def _log_message(self, message: str, level: str = "info", log_current_run: bool = True, error_details: str = ""):
+        """Log a message using the configured logging function."""
+        self._log_function(
+            logger=self.logger,
+            message=message,
+            error_details=error_details,
+            level=level,
+            log_current_run=log_current_run,
+            exception_class=PusherCacheError,
+        )