From a9a51de58d91d706581507cc7309f097a1ea17a3 Mon Sep 17 00:00:00 2001 From: mcarans Date: Thu, 15 Oct 2026 14:35:23 +1300 Subject: [PATCH] Add DataPusher+ submission and data dictionary update to Resource - Resource.submit_to_datapusher (datapusher_submit) and Resource.get_datapusher_status (datapusher_status) - Resource.update_hdx_data_dictionary: merges into or replaces hdx_data_dictionary via resource_patch, then updates datastore field labels/descriptions via datastore_create without reloading data. last_modified is unchanged so no dataset update notifications are sent. --- documentation/index.md | 32 ++++++ src/hdx/data/resource.py | 103 ++++++++++++++++++++ tests/hdx/data/test_resource.py | 167 ++++++++++++++++++++++++++++++++ 3 files changed, 302 insertions(+) diff --git a/documentation/index.md b/documentation/index.md index de6bd3e..bc6daf6 100755 --- a/documentation/index.md +++ b/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` diff --git a/src/hdx/data/resource.py b/src/hdx/data/resource.py index f5dbb57..9fd4c41 100755 --- a/src/hdx/data/resource.py +++ b/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 diff --git a/tests/hdx/data/test_resource.py b/tests/hdx/data/test_resource.py index d7002d9..6229422 100755 --- a/tests/hdx/data/test_resource.py +++ b/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 (