HDXTABLE-147 Add datapusher+ and dictionary methods to HDX Python API

documentation/index.md

@@ -991,6 +991,27 @@ using the getter:
 
     data_dictionary = resource.get_hdx_data_dictionary()
 
+To change the data dictionary of a resource already in HDX without sending
+dataset update notifications, use **update_hdx_data_dictionary**. It patches
+only the `hdx_data_dictionary` field (so `last_modified` is unchanged), then,
+if the resource has a datastore, updates the datastore field labels and
+descriptions without reloading the data. The given columns are merged by
+`field` into the existing data dictionary in the resource metadata, so read
+the resource from HDX first. Pass `replace=True` to replace the data
+dictionary instead and `update_datastore=False` to leave the datastore alone:
+
+    resource = Resource.read_from_hdx(RESOURCE_ID)
+    resource.update_hdx_data_dictionary([
+        {
+            "field": "col_name",
+            "label": "Column Label",
+            "description": "New description.",
+        },
+    ])
+
+Columns are matched to datastore fields by `field`. A warning is logged for
+any column with no matching datastore field.
+
 ### Datastore
 
 A resource's data can be loaded into the HDX datastore and queried. HDX calls this
@@ -1009,6 +1030,17 @@ the datastore for a resource:
     has_datastore = resource.has_datastore()
     resource.delete_datastore()
 
+HDX loads uploaded files into the datastore using DataPusher+. To reload a
+resource's file into the datastore without uploading it again (so
+`last_modified` is unchanged), use **submit_to_datapusher**. It returns False
+if the job was not submitted, eg. because one is already pending. HDX's
+datastore allowlist and format checks, which apply to automatic submissions
+on upload, are not applied. **get_datapusher_status** returns the status of
+the latest job or None if there is none:
+
+    submitted = resource.submit_to_datapusher(ignore_hash=True)
+    status = resource.get_datapusher_status()  # eg. status["status"] == "complete"
+
 To download the whole datastore (rather than the resource's file) use
 **download_datastore**, which returns the URL and path like **download**.
 `file_format` can be `csv` (default), `tsv`, `json` or `xml`, and `filename`

src/hdx/data/resource.py

@@ -72,6 +72,8 @@ def actions() -> dict[str, str]:
             "datastore_upsert": "datastore_upsert",
             "datastore_search": "datastore_search",
             "datastore_search_sql": "datastore_search_sql",
+            "datapusher_submit": "datapusher_submit",
+            "datapusher_status": "datapusher_status",
         }
 
     def update_from_yaml(
@@ -319,6 +321,72 @@ def set_hdx_data_dictionary(self, data_dictionary: Sequence[dict]) -> None:
             data_dictionary, separators=(",", ":")
         )
 
+    def update_hdx_data_dictionary(
+        self,
+        data_dictionary: Sequence[dict],
+        replace: bool = False,
+        update_datastore: bool = True,
+    ) -> None:
+        """Update the resource's data dictionary (column definitions) in HDX
+        without changing last_modified, so no dataset update notifications are
+        sent. Column definitions are as for set_hdx_data_dictionary. Unless
+        replace is True, they are merged by field into the existing data
+        dictionary in the resource metadata (so call read_from_hdx first).
+        The resource's hdx_data_dictionary field is patched, then if
+        update_datastore is True and the resource has a datastore, the
+        datastore field labels and descriptions are updated using
+        datastore_create without reloading the data.
+
+        Args:
+            data_dictionary: Sequence of column definition dictionaries
+            replace: Replace rather than merge into existing data dictionary. Defaults to False.
+            update_datastore: Update datastore field labels and descriptions. Defaults to True.
+
+        Returns:
+            None
+        """
+        if replace:
+            merged = [dict(column) for column in data_dictionary]
+        else:
+            new_columns = {column["field"]: column for column in data_dictionary}
+            merged = []
+            for column in self.get_hdx_data_dictionary() or []:
+                new_column = new_columns.pop(column.get("field"), None)
+                if new_column:
+                    merged.append({**column, **new_column})
+                else:
+                    merged.append(column)
+            merged.extend(new_columns.values())
+        self.set_hdx_data_dictionary(merged)
+        self._write_to_hdx(
+            "patch",
+            {
+                "id": self.data["id"],
+                "hdx_data_dictionary": self.data["hdx_data_dictionary"],
+            },
+            "id",
+        )
+        if not update_datastore or not self.has_datastore():
+            return
+        columns = {column["field"]: column for column in merged}
+        fields = []
+        for field in self.get_datastore_fields():
+            info = dict(field.get("info") or {})
+            column = columns.pop(field["id"], None)
+            if column:
+                info["label"] = column["label"]
+                info["notes"] = column["description"]
+            fields.append({"id": field["id"], "type": field["type"], "info": info})
+        if columns:
+            logger.warning(
+                f"Data dictionary fields {', '.join(columns)} not in datastore for resource {self.data['id']}!"
+            )
+        self._write_to_hdx(
+            "datastore_create",
+            {"resource_id": self.data["id"], "force": True, "fields": fields},
+            "resource_id",
+        )
+
     def clean_format(self) -> str:
         """Clean the resource's format, setting it to None if it is invalid and
         cannot be mapped
@@ -1015,6 +1083,41 @@ def delete_datastore(self) -> None:
         if not success:
             logger.debug(result)
 
+    def submit_to_datapusher(self, ignore_hash: bool = False) -> bool:
+        """Submit the resource to DataPusher+ to load its file into the
+        datastore. This does not change last_modified. Unlike the automatic
+        submission on file upload, HDX's datastore allowlist and format checks
+        are not applied.
+
+        Args:
+            ignore_hash: Reload even if the file hash is unchanged. Defaults to False.
+
+        Returns:
+            True if submitted, False if not (eg. a job is already pending)
+        """
+        return self._write_to_hdx(
+            "datapusher_submit",
+            {"resource_id": self.data["id"], "ignore_hash": ignore_hash},
+            "resource_id",
+        )
+
+    def get_datapusher_status(self) -> dict | None:
+        """Get the status of the resource's latest DataPusher+ job
+
+        Returns:
+            Status dictionary with keys including status and error or None if no job
+        """
+        success, result = self._read_from_hdx(
+            "datapusher",
+            self.data["id"],
+            "resource_id",
+            self.actions()["datapusher_status"],
+        )
+        if success:
+            return result
+        logger.debug(result)
+        return None
+
     def get_resource_views(self) -> list[ResourceView]:
         """Get any resource views in the resource
 

tests/hdx/data/test_resource.py

@@ -747,6 +747,70 @@ def post(url, data, **kwargs):
 
         Configuration.read().remoteckan().session = MockSession()
 
+    @pytest.fixture(scope="function")
+    def post_datapusher(self):
+        calls = []
+
+        class MockSession:
+            @staticmethod
+            def post(url, data, **kwargs):
+                datadict = json.loads(data.decode("utf-8"))
+                action = url.split("/")[-1]
+                calls.append((action, datadict))
+                if datadict.get("resource_id") == "NOTFOUND":
+                    return MockResponse(
+                        404,
+                        f'{{"success": false, "error": {{"message": "Not found", "__type": "Not Found Error"}}, "help": "http://test-data.humdata.org/api/3/action/help_show?name={action}"}}',
+                    )
+                if action == "datapusher_submit":
+                    result = "true"
+                elif action == "datapusher_status":
+                    result = '{"status": "complete", "error": {}}'
+                elif action == "resource_patch":
+                    resultdictcopy = copy.deepcopy(resultdict)
+                    resultdictcopy["hdx_data_dictionary"] = datadict[
+                        "hdx_data_dictionary"
+                    ]
+                    result = json.dumps(resultdictcopy)
+                elif action == "datastore_search":
+                    result = json.dumps(
+                        {
+                            "fields": [
+                                {"id": "_id", "type": "int"},
+                                {
+                                    "id": "code",
+                                    "type": "text",
+                                    "info": {
+                                        "label": "Code",
+                                        "notes": "Old code",
+                                        "type_override": "text",
+                                    },
+                                },
+                                {
+                                    "id": "value",
+                                    "type": "numeric",
+                                    "info": {"label": "Value", "notes": "Old value"},
+                                },
+                                {"id": "notes", "type": "text"},
+                            ],
+                            "records": [],
+                        }
+                    )
+                elif action == "datastore_create":
+                    result = json.dumps(datadict)
+                else:
+                    return MockResponse(
+                        404,
+                        f'{{"success": false, "error": {{"message": "TEST ERROR: Unexpected {action}", "__type": "TEST ERROR"}}, "help": "http://test-data.humdata.org/api/3/action/help_show?name={action}"}}',
+                    )
+                return MockResponse(
+                    200,
+                    f'{{"success": true, "result": {result}, "help": "http://test-data.humdata.org/api/3/action/help_show?name={action}"}}',
+                )
+
+        Configuration.read().remoteckan().session = MockSession()
+        return calls
+
     def test_read_from_hdx(self, configuration, read):
         resource = Resource.read_from_hdx("74b74ae1-df0c-4716-829f-4f939a046811")
         assert resource["id"] == "de6549d8-268b-4dfe-adaf-a4ae5c8510d5"
@@ -1343,6 +1407,109 @@ def test_get_fs_check_info(self, configuration):
         resource["fs_check_info"] = events
         assert resource.get_fs_check_info() == events
 
+    def test_datapusher(self, configuration, post_datapusher):
+        resource = Resource({"id": "de6549d8-268b-4dfe-adaf-a4ae5c8510d5"})
+        assert resource.submit_to_datapusher(ignore_hash=True) is True
+        assert post_datapusher[-1] == (
+            "datapusher_submit",
+            {
+                "resource_id": "de6549d8-268b-4dfe-adaf-a4ae5c8510d5",
+                "ignore_hash": True,
+            },
+        )
+        assert resource.get_datapusher_status() == {
+            "status": "complete",
+            "error": {},
+        }
+        resource = Resource({"id": "NOTFOUND"})
+        assert resource.get_datapusher_status() is None
+        with pytest.raises(HDXError):
+            resource.submit_to_datapusher()
+
+    def test_update_hdx_data_dictionary(self, configuration, post_datapusher):
+        existing = [
+            {
+                "field": "code",
+                "label": "Code",
+                "description": "Old code",
+                "data_type": "text",
+            },
+            {"field": "value", "label": "Value", "description": "Old value"},
+        ]
+        resource = Resource(
+            {
+                "id": "de6549d8-268b-4dfe-adaf-a4ae5c8510d5",
+                "hdx_data_dictionary": json.dumps(existing),
+            }
+        )
+        resource.update_hdx_data_dictionary(
+            [
+                {"field": "code", "label": "Code", "description": "New code"},
+                {"field": "extra", "label": "Extra", "description": "Not in datastore"},
+            ]
+        )
+        expected = [
+            {
+                "field": "code",
+                "label": "Code",
+                "description": "New code",
+                "data_type": "text",
+            },
+            {"field": "value", "label": "Value", "description": "Old value"},
+            {"field": "extra", "label": "Extra", "description": "Not in datastore"},
+        ]
+        assert resource.get_hdx_data_dictionary() == expected
+        actions = [action for action, _ in post_datapusher]
+        assert actions == [
+            "resource_patch",
+            "datastore_search",
+            "datastore_search",
+            "datastore_create",
+        ]
+        _, patch = post_datapusher[0]
+        assert patch == {
+            "id": "de6549d8-268b-4dfe-adaf-a4ae5c8510d5",
+            "hdx_data_dictionary": json.dumps(expected, separators=(",", ":")),
+        }
+        _, create = post_datapusher[-1]
+        assert create == {
+            "resource_id": "de6549d8-268b-4dfe-adaf-a4ae5c8510d5",
+            "force": True,
+            "fields": [
+                {
+                    "id": "code",
+                    "type": "text",
+                    "info": {
+                        "label": "Code",
+                        "notes": "New code",
+                        "type_override": "text",
+                    },
+                },
+                {
+                    "id": "value",
+                    "type": "numeric",
+                    "info": {"label": "Value", "notes": "Old value"},
+                },
+                {"id": "notes", "type": "text", "info": {}},
+            ],
+        }
+
+        post_datapusher.clear()
+        resource.update_hdx_data_dictionary(
+            [{"field": "value", "label": "Value", "description": "New value"}],
+            replace=True,
+            update_datastore=False,
+        )
+        assert resource.get_hdx_data_dictionary() == [
+            {"field": "value", "label": "Value", "description": "New value"}
+        ]
+        assert [action for action, _ in post_datapusher] == ["resource_patch"]
+
+        with pytest.raises(HDXError):
+            resource.update_hdx_data_dictionary(
+                [{"field": "value", "label": "Value"}], replace=True
+            )
+
     def test_get_hdx_url(self, configuration, read):
         resource = Resource.read_from_hdx("74b74ae1-df0c-4716-829f-4f939a046811")
         assert (