From 2ed2b6ffe312a3b7902909fb743814f6a418c7af Mon Sep 17 00:00:00 2001 From: mcarans Date: Tue, 13 Oct 2026 12:36:26 +1300 Subject: [PATCH] Add datastore query and dump helpers to Resource Add search_datastore, get_datastore_records, get_datastore_fields, search_datastore_sql, get_datastore_dump_url and download_datastore. Document datastore methods. --- documentation/index.md | 52 +++++++++ src/hdx/data/resource.py | 181 +++++++++++++++++++++++++++++++- tests/hdx/data/test_resource.py | 164 +++++++++++++++++++++++++++++ 3 files changed, 396 insertions(+), 1 deletion(-) diff --git a/documentation/index.md b/documentation/index.md index 12b12a1..fb173ac 100755 --- a/documentation/index.md +++ b/documentation/index.md @@ -965,6 +965,58 @@ using the getter: data_dictionary = resource.get_hdx_data_dictionary() +### Datastore + +A resource's data can be loaded into the HDX datastore and queried. HDX +requires an API key for all datastore actions and dumps, including reads of +public resources. To create, update or delete the datastore for a resource: + + resource.create_datastore( + [{"id": "code", "type": "text"}, {"id": "value", "type": "numeric"}], + primary_key="code", + ) + resource.update_datastore([{"code": "A", "value": 1.0}]) # upsert by default + has_datastore = resource.has_datastore() + resource.delete_datastore() + +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` +defaults to the resource name with that extension. **get_datastore_dump_url** +returns the URL without downloading: + + url, path = resource.download_datastore("FOLDER_TO_DOWNLOAD_TO") + url = resource.get_datastore_dump_url(file_format="json") + +**search_datastore** wraps CKAN's `datastore_search`, returning the result +dictionary (with keys including `records`, `fields` and `total`). It accepts +`filters`, `q`, `fields`, `sort`, `distinct`, `limit` (default 100), `offset` +and `include_total`, and passes any other keyword arguments to CKAN: + + result = resource.search_datastore( + filters={"code": ["A", "B"]}, fields=["code", "value"], sort="value desc" + ) + records = result["records"] + +**get_datastore_records** takes the same query parameters but pages through +all matching rows, yielding each record (`page_size` defaults to 10000): + + for record in resource.get_datastore_records(filters={"code": "A"}): + ... + +**get_datastore_fields** returns the datastore field definitions excluding the +internal `_id` field: + + fields = resource.get_datastore_fields() + +**search_datastore_sql** wraps CKAN's `datastore_search_sql`, running a +SELECT statement where tables are named by double quoted resource id and +returning the list of records: + + records = Resource.search_datastore_sql( + 'SELECT code, SUM(value) AS total FROM "RESOURCE_ID" GROUP BY code' + ) + ## Showcase Management The **Showcase** class enables you to manage showcases, creating, deleting and updating diff --git a/src/hdx/data/resource.py b/src/hdx/data/resource.py index 40b0ca9..4a9ba7c 100755 --- a/src/hdx/data/resource.py +++ b/src/hdx/data/resource.py @@ -2,7 +2,7 @@ import json import logging -from collections.abc import Sequence +from collections.abc import Iterator, Sequence from datetime import datetime from pathlib import Path from typing import Any, Optional @@ -34,6 +34,7 @@ class Resource(HDXObject): """ _formats_dict = None + _valid_datastore_dump_formats = frozenset({"csv", "tsv", "json", "xml"}) _valid_data_dictionary_types = frozenset( {"text", "numeric", "date", "timestamp without time zone"} ) @@ -70,6 +71,7 @@ def actions() -> dict[str, str]: "datastore_insert": "datastore_insert", "datastore_upsert": "datastore_upsert", "datastore_search": "datastore_search", + "datastore_search_sql": "datastore_search_sql", } def update_from_yaml( @@ -711,6 +713,62 @@ def download( path = downloader.download_file(url, folder=folder, filename=filename) return url, path + def get_datastore_dump_url(self, file_format: str = "csv") -> str: + """Get the URL from which the resource datastore can be downloaded in + full. + + Args: + file_format: Dump format (csv, tsv, json or xml). Defaults to csv. + + Returns: + Datastore dump URL + """ + file_format = file_format.lower() + if file_format not in self._valid_datastore_dump_formats: + valid_formats = ", ".join(sorted(self._valid_datastore_dump_formats)) + raise HDXError( + f"Invalid datastore dump format '{file_format}'! Must be one of: {valid_formats}" + ) + url = ( + f"{self.configuration.get_hdx_site_url()}/datastore/dump/{self.data['id']}" + ) + if file_format != "csv": + url = f"{url}?format={file_format}" + return url + + def download_datastore( + self, + folder: Path | str | None = None, + filename: str | None = None, + file_format: str = "csv", + retriever: Retrieve | None = None, + ) -> tuple[str, Path]: + """Download the resource datastore in full to provided folder or + temporary folder if no folder supplied + + Args: + folder: Folder to download datastore to. Defaults to None. + filename: Filename to use. Defaults to None (resource name with file_format extension). + file_format: Dump format (csv, tsv, json or xml). Defaults to csv. + retriever: Retrieve object to use. Defaults to None. + + Returns: + (URL downloaded, Path to downloaded file) + """ + url = self.get_datastore_dump_url(file_format) + logger.debug(f"Downloading {url}") + if filename is None: + filename = self.data["name"] + extension = f".{file_format.lower()}" + if not filename.endswith(extension): + filename = f"{filename}{extension}" + session = hdx.api.utilities.url_utils.get_ckan_ready_session(self.configuration) + with Download(session=session) as downloader: + if retriever: + downloader = retriever.clone(downloader) + path = downloader.download_file(url, folder=folder, filename=filename) + return url, path + @staticmethod def get_all_resource_ids_in_datastore( configuration: Configuration | None = None, @@ -805,6 +863,127 @@ def update_datastore( } self._write_to_hdx("datastore_upsert", data, "resource_id") + def search_datastore( + self, + filters: dict | None = None, + q: str | dict | None = None, + fields: Sequence[str] | None = None, + sort: str | None = None, + distinct: bool = False, + limit: int = 100, + offset: int = 0, + include_total: bool = True, + **kwargs: Any, + ) -> dict: + """Search the resource datastore using CKAN datastore_search. + + Args: + filters: Dictionary of field name to value(s) to match. Defaults to None. + q: Full text query string or dictionary of field name to query. Defaults to None. + fields: Fields to return. Defaults to None (all fields). + sort: Comma separated field names with optional asc/desc eg. "code desc". Defaults to None. + distinct: Whether to return only distinct rows. Defaults to False. + limit: Maximum number of rows to return. Defaults to 100. + offset: Number of rows to skip. Defaults to 0. + include_total: Whether to include total number of matching rows. Defaults to True. + **kwargs: Other parameters to pass to datastore_search eg. plain, language. + + Returns: + datastore_search result dictionary with keys including records, fields and total + """ + data: dict = { + "resource_id": self.data["id"], + "limit": limit, + "offset": offset, + "include_total": include_total, + } + if filters is not None: + data["filters"] = filters + if q is not None: + data["q"] = q + if fields is not None: + data["fields"] = ",".join(fields) + if sort is not None: + data["sort"] = sort + if distinct: + data["distinct"] = True + data.update(kwargs) + return self._write_to_hdx("datastore_search", data, "resource_id") + + def get_datastore_records( + self, + filters: dict | None = None, + q: str | dict | None = None, + fields: Sequence[str] | None = None, + sort: str | None = None, + distinct: bool = False, + page_size: int = 10000, + **kwargs: Any, + ) -> Iterator[dict]: + """Iterate over all matching records in the resource datastore, paging + through datastore_search results. + + Args: + filters: Dictionary of field name to value(s) to match. Defaults to None. + q: Full text query string or dictionary of field name to query. Defaults to None. + fields: Fields to return. Defaults to None (all fields). + sort: Comma separated field names with optional asc/desc eg. "code desc". Defaults to None. + distinct: Whether to return only distinct rows. Defaults to False. + page_size: Number of rows to request per call. Defaults to 10000. + **kwargs: Other parameters to pass to datastore_search eg. plain, language. + + Returns: + Iterator of record dictionaries + """ + offset = 0 + while True: + result = self.search_datastore( + filters=filters, + q=q, + fields=fields, + sort=sort, + distinct=distinct, + limit=page_size, + offset=offset, + include_total=False, + **kwargs, + ) + records = result["records"] + yield from records + if len(records) < page_size: + return + offset += page_size + + def get_datastore_fields(self) -> list[dict]: + """Get the field definitions of the resource datastore excluding the + internal _id field. + + Returns: + List of field dictionaries each with id and type keys + """ + result = self.search_datastore(limit=0, include_total=False) + return [field for field in result["fields"] if field["id"] != "_id"] + + @staticmethod + def search_datastore_sql( + sql: str, + configuration: Configuration | None = None, + ) -> list[dict]: + """Run a read-only SQL query against the HDX datastore using CKAN + datastore_search_sql. Tables are named by resource id, which must be + double quoted eg. SELECT * FROM "RESOURCE_ID". + + Args: + sql: SQL SELECT statement + configuration: HDX configuration. Defaults to global configuration. + + Returns: + List of record dictionaries + """ + resource = Resource(configuration=configuration) + result = resource._write_to_hdx("datastore_search_sql", {"sql": sql}) + return result["records"] + def delete_datastore(self) -> None: """Delete a resource from the HDX datastore diff --git a/tests/hdx/data/test_resource.py b/tests/hdx/data/test_resource.py index b766e82..6c7fb47 100755 --- a/tests/hdx/data/test_resource.py +++ b/tests/hdx/data/test_resource.py @@ -6,6 +6,7 @@ from datetime import datetime, timezone from os import remove from os.path import basename +from pathlib import Path import pytest from hdx.utilities.dateparse import parse_date @@ -338,6 +339,7 @@ def mockresourceview(url, decodedata): class TestResource: datastore = None + datastore_request = None @pytest.fixture(scope="class") def static_yaml(self, configfolder): @@ -600,6 +602,73 @@ def post(url, data, **kwargs): Configuration.read().remoteckan().session = MockSession() + @pytest.fixture(scope="function") + def post_datastore_query(self): + all_records = [ + {"_id": 1, "code": "A", "value": 1.0}, + {"_id": 2, "code": "B", "value": 2.0}, + {"_id": 3, "code": "C", "value": 3.0}, + ] + fields = [ + {"id": "_id", "type": "int"}, + {"id": "code", "type": "text"}, + {"id": "value", "type": "numeric"}, + ] + + class MockSession: + @staticmethod + def post(url, data, **kwargs): + datadict = json.loads(data.decode("utf-8")) + if url.endswith("datastore_search_sql"): + if "error" in datadict["sql"]: + return MockResponse( + 409, + '{"success": false, "error": {"message": "Bad SQL", "__type": "Validation Error"}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=datastore_search_sql"}', + ) + result = {"records": all_records[:2], "fields": fields} + return MockResponse( + 200, + json.dumps( + { + "success": True, + "result": result, + "help": "http://test-data.humdata.org/api/3/action/help_show?name=datastore_search_sql", + } + ), + ) + if url.endswith("datastore_search"): + if datadict["resource_id"] != "datastore_query_resource": + return MockResponse( + 404, + '{"success": false, "error": {"message": "Not found", "__type": "Not Found Error"}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=datastore_search"}', + ) + TestResource.datastore_request = datadict + offset = datadict["offset"] + limit = datadict["limit"] + result = { + "resource_id": datadict["resource_id"], + "fields": fields, + "records": all_records[offset : offset + limit], + } + if datadict["include_total"]: + result["total"] = len(all_records) + return MockResponse( + 200, + json.dumps( + { + "success": True, + "result": result, + "help": "http://test-data.humdata.org/api/3/action/help_show?name=datastore_search", + } + ), + ) + return MockResponse( + 404, + '{"success": false, "error": {"message": "TEST ERROR: Not datastore search", "__type": "TEST ERROR: Not Datastore Search Error"}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=datastore_search"}', + ) + + Configuration.read().remoteckan().session = MockSession() + @pytest.fixture(scope="function") def post_patch(self): class MockSession: @@ -1031,6 +1100,39 @@ def test_download(self, configuration, read): with pytest.raises(DownloadError): resource2.download() + def test_download_datastore(self, configuration, read, monkeypatch): + downloads = [] + + def download_file(self, url, folder=None, filename=None, path=None, **kwargs): + if path is None: + path = Path(folder or "tmp") / filename + downloads.append(url) + return path + + monkeypatch.setattr(Download, "download_file", download_file) + resource = Resource.read_from_hdx("74b74ae1-df0c-4716-829f-4f939a046811") + site_url = resource.configuration.get_hdx_site_url() + dump_url = f"{site_url}/datastore/dump/{resource['id']}" + assert resource.get_datastore_dump_url() == dump_url + assert resource.get_datastore_dump_url("JSON") == f"{dump_url}?format=json" + with pytest.raises(HDXError): + resource.get_datastore_dump_url("xlsx") + + url, path = resource.download_datastore("folder") + assert url == dump_url + assert path == Path("folder", "MyResource1.csv") + url, path = resource.download_datastore( + filename="builds.tsv", file_format="tsv" + ) + assert url == f"{dump_url}?format=tsv" + assert path == Path("tmp", "builds.tsv") + tempdir = get_temp_dir() + retriever = Retrieve(Download(user_agent="test"), tempdir, tempdir, tempdir) + url, path = resource.download_datastore(retriever=retriever) + assert url == dump_url + assert path.name == "MyResource1.csv" + assert len(downloads) == 3 + def test_datastore(self, configuration, post_datastore, topline_yaml, topline_json): resource_ids = Resource.get_all_resource_ids_in_datastore() assert resource_ids == [ @@ -1073,6 +1175,68 @@ def test_datastore(self, configuration, post_datastore, topline_yaml, topline_js resource.update_datastore(records, method="insert") assert TestResource.datastore == "upsert" + def test_datastore_queries(self, configuration, post_datastore_query): + resource = Resource({"id": "datastore_query_resource"}) + result = resource.search_datastore() + assert result["total"] == 3 + assert len(result["records"]) == 3 + assert TestResource.datastore_request == { + "resource_id": "datastore_query_resource", + "limit": 100, + "offset": 0, + "include_total": True, + } + + resource.search_datastore( + filters={"code": ["A", "B"]}, + q="A", + fields=["code", "value"], + sort="value desc", + distinct=True, + limit=2, + offset=1, + plain=False, + ) + assert TestResource.datastore_request == { + "resource_id": "datastore_query_resource", + "limit": 2, + "offset": 1, + "include_total": True, + "filters": {"code": ["A", "B"]}, + "q": "A", + "fields": "code,value", + "sort": "value desc", + "distinct": True, + "plain": False, + } + + records = list(resource.get_datastore_records(sort="code", page_size=2)) + assert [record["code"] for record in records] == ["A", "B", "C"] + assert TestResource.datastore_request["offset"] == 2 + assert TestResource.datastore_request["sort"] == "code" + assert TestResource.datastore_request["include_total"] is False + + records = list(resource.get_datastore_records(page_size=3)) + assert len(records) == 3 + assert TestResource.datastore_request["offset"] == 3 + + assert resource.get_datastore_fields() == [ + {"id": "code", "type": "text"}, + {"id": "value", "type": "numeric"}, + ] + assert TestResource.datastore_request["limit"] == 0 + + records = Resource.search_datastore_sql( + 'SELECT * FROM "datastore_query_resource" LIMIT 2' + ) + assert [record["code"] for record in records] == ["A", "B"] + with pytest.raises(HDXError): + Resource.search_datastore_sql("error") + + resource = Resource({"id": "datastore_unknown_resource"}) + with pytest.raises(HDXError): + resource.search_datastore() + def test_resource_views(self, configuration, post_resourceview): resource = Resource({"id": "25982d1c-f45a-45e1-b14e-87d367413045"}) with pytest.raises(HDXError):