Add wrappers for HDX data grid, resource flag, QA, fs_check and datastore actions

documentation/index.md

@@ -647,6 +647,35 @@ This returns a list of dictionaries, each from a resource name to a list of
 check events with keys such as `state`, `timestamp`, `sheet_changes` and
 `hxl_proxy_response`.
 
+Similarly, you can get the geopreview shape information of a dataset's
+resources:
+
+    shape_info = dataset.get_shape_info()
+
+HDX can recommend tags for a dataset based on its title and organization
+(`owner_org`). Tags are ordered from most to least recommended:
+
+    tags = dataset.get_recommended_tags()
+
+You can check if the dataset or its organization is in HDX's allowlist for
+automatic loading of resources into the datastore:
+
+    allowed = dataset.is_allowed_for_datastore()
+
+HDX sysadmins can submit all of a dataset's CSV resources to DataPusher+
+(returning False if there are none):
+
+    submitted = dataset.push_to_datastore()
+
+Users who are sysadmins or have permission to manage QA can mark QA of a
+dataset as completed, set a QA field of all its resources or clear a file
+structure check field (`fs_check_info` by default) of all its resources. The
+dataset is updated from the result:
+
+    dataset.set_qa_completed(True)
+    dataset.qa_revise_resources("pii", "false")
+    dataset.reset_fs_check()
+
 ### Time Period
 
 Time Period is a mandatory field in HDX. It is the earliest start date and
@@ -964,6 +993,32 @@ HDX:
 
     events = resource.get_fs_check_info()
 
+Users who are sysadmins or have permission to manage QA can set or clear a
+file structure check field of a resource. If the key is `fs_check_info` and the
+value is a dictionary, it is appended to the existing check events (the last 20
+are kept):
+
+    resource.revise_fs_check("fs_check_info", {"state": "success", ...})
+    resource.reset_fs_check()  # key defaults to fs_check_info
+
+They can also quarantine a resource (which removes its `shape_info` and
+geopreview) or take it out of quarantine, and patch other QA fields:
+
+    in_quarantine = resource.is_in_quarantine()
+    resource.set_in_quarantine(True)
+    resource.qa_patch(pii="true", sdc_report_flag="false")
+
+A resource can be flagged as containing p-codes or as being in HAPI. Setting
+these flags requires sysadmin or permission to manage p-codes or the in HAPI
+flag respectively:
+
+    p_coded = resource.is_p_coded()  # None if not set
+    resource.set_p_coded(True)
+    in_hapi = resource.is_in_hapi()
+    resource.set_in_hapi(True)
+
+These methods update the resource from the result.
+
 ### Data Dictionary
 
 A resource can have a data dictionary describing the columns in the data it
@@ -1041,6 +1096,11 @@ 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"
 
+HDX sysadmins can instead use **push_to_datastore**, which submits the
+resource through HDX's wrapper action:
+
+    submitted = resource.push_to_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`
@@ -1213,6 +1273,17 @@ since Data Grid membership changes over time, rather than a fixed list):
 
 This returns a sorted list of 3-letter location codes.
 
+HDX sysadmins can get the Data Grid of a country and clear HDX's cached copy of
+it:
+
+    data_grid = Location.get_data_grid("AFG")
+    Location.invalidate_data_grid("AFG")
+
+**get_data_grid** returns a dictionary with keys including `title`, `date`,
+`stats` and `categories` (or None if not found). Each category has
+`subcategories`, each with a list of `datasets`. Complementary datasets have
+`is_complementary` set to True.
+
 You can get the datasets belonging to a location as follows:
 
     datasets = location.get_datasets(**kwargs)
@@ -1298,6 +1369,14 @@ and push them to analytics:
 
     pushed_stats = push_general_stats()
 
+HDX sysadmins can get the number of activities of each active user (optionally
+between start and end dates) and lists of active and inactive organizations:
+
+    from hdx.api.utilities.hdx_stats import get_organization_statistics, get_user_statistics
+
+    user_stats = get_user_statistics(start_date="2026-01-01", end_date="2026-06-30")
+    org_stats = get_organization_statistics()
+
 ## Pipeline State
 
 The HDXState class allows the reading and writing of state to a given dataset. Input

src/hdx/api/utilities/hdx_stats.py

@@ -4,20 +4,23 @@
 from hdx.data.hdxobject import HDXError
 
 
-def _call_stats_action(action: str, configuration: Configuration | None) -> dict:
+def _call_stats_action(
+    action: str, configuration: Configuration | None, data: dict | None = None
+) -> dict:
     """Call an HDX statistics action
 
     Args:
         action: CKAN action name
         configuration: HDX configuration. Defaults to global configuration.
+        data: Parameters to pass to the action. Defaults to None.
 
     Returns:
         Statistics dictionary
     """
     if configuration is None:
         configuration = Configuration.read()
     try:
-        return configuration.call_remoteckan(action, {})
+        return configuration.call_remoteckan(action, data or {})
     except Exception as e:
         raise HDXError(f"Failed when trying to call {action}!") from e
 
@@ -45,3 +48,39 @@ def push_general_stats(configuration: Configuration | None = None) -> dict:
         Statistics that were pushed
     """
     return _call_stats_action("hdx_push_general_stats", configuration)
+
+
+def get_user_statistics(
+    start_date: str | None = None,
+    end_date: str | None = None,
+    configuration: Configuration | None = None,
+) -> dict:
+    """Get the number of activities of each active user on HDX, optionally
+    between start and end dates. Requires sysadmin.
+
+    Args:
+        start_date: Count activities on or after this date eg. 2026-01-01. Defaults to None.
+        end_date: Count activities on or before this date. Defaults to None.
+        configuration: HDX configuration. Defaults to global configuration.
+
+    Returns:
+        Dictionary from user id to dictionary with keys id, name and no_of_activities
+    """
+    data = {}
+    if start_date:
+        data["start_date"] = start_date
+    if end_date:
+        data["end_date"] = end_date
+    return _call_stats_action("hdx_user_statistics", configuration, data)
+
+
+def get_organization_statistics(configuration: Configuration | None = None) -> dict:
+    """Get active and inactive organizations on HDX. Requires sysadmin.
+
+    Args:
+        configuration: HDX configuration. Defaults to global configuration.
+
+    Returns:
+        Dictionary with keys active_organizations and inactive_organizations
+    """
+    return _call_stats_action("hdx_organization_statistics", configuration)

src/hdx/data/dataset.py

@@ -148,6 +148,13 @@ def actions() -> dict[str, str]:
             "autocomplete": "package_autocomplete",
             "create_default_views": "package_create_default_resource_views",
             "fs_check_info": "fs_check_info_show",
+            "fs_check_reset": "hdx_fs_check_package_reset",
+            "shape_info": "shape_info_show",
+            "recommend_tags": "hdx_recommend_tags",
+            "allowed_for_datastore": "hdx_is_package_allowed_for_datastore",
+            "push_to_datastore": "hdx_push_resource_to_datastore",
+            "qa_completed": "hdx_mark_qa_completed",
+            "qa_revise_resources": "hdx_qa_package_revise_resource",
         }
 
     def __setitem__(self, key: Any, value: Any) -> None:
@@ -2347,6 +2354,136 @@ def get_fs_check_info(self) -> list[dict]:
             return result
         return []
 
+    def _set_from_package(self, package: dict) -> None:
+        """Set the dataset's metadata and resources from a package metadata
+        dictionary returned by HDX
+
+        Args:
+            package: Package metadata dictionary
+
+        Returns:
+            None
+        """
+        self.data = package
+        self.init_resources()
+        self.separate_resources()
+
+    def reset_fs_check(self, key: str = "fs_check_info") -> None:
+        """Clear a file structure check field of all the dataset's resources
+        in HDX. Requires sysadmin or permission to manage QA.
+
+        Args:
+            key: Resource field to clear. Defaults to fs_check_info.
+
+        Returns:
+            None
+        """
+        result = self._write_to_hdx(
+            "fs_check_reset", {"package_id": self.data["id"], "key": key}, "package_id"
+        )
+        self._set_from_package(result["package"])
+
+    def set_qa_completed(self, qa_completed: bool) -> None:
+        """Mark QA of the dataset in HDX as completed or not. Requires
+        sysadmin or permission to manage QA.
+
+        Args:
+            qa_completed: Whether QA is completed
+
+        Returns:
+            None
+        """
+        result = self._write_to_hdx(
+            "qa_completed", {"id": self.data["id"], "qa_completed": qa_completed}, "id"
+        )
+        self._set_from_package(result["package"])
+
+    def qa_revise_resources(self, key: str, value: Any) -> None:
+        """Set a QA field of all the dataset's resources in HDX. Requires
+        sysadmin or permission to manage QA.
+
+        Args:
+            key: Resource field eg. pii
+            value: Value to set
+
+        Returns:
+            None
+        """
+        result = self._write_to_hdx(
+            "qa_revise_resources",
+            {"id": self.data["id"], "key": key, "value": value},
+            "id",
+        )
+        self._set_from_package(result["package"])
+
+    def get_shape_info(self) -> list[dict]:
+        """Get geopreview shape information for the dataset's resources from
+        HDX
+
+        Returns:
+            List of dictionaries from resource name to shape information
+        """
+        success, result = self._read_from_hdx(
+            "dataset",
+            self.get_name_or_id(prefer_name=False),
+            "id",
+            self.actions()["shape_info"],
+        )
+        if success:
+            return result
+        return []
+
+    def get_recommended_tags(self) -> list[str]:
+        """Get tags recommended by HDX for the dataset based on its title and
+        organization. Tags are ordered from most to least recommended.
+
+        Returns:
+            List of recommended tags
+        """
+        data = {"title": self.data.get("title")}
+        organization = self.data.get("owner_org")
+        if organization:
+            data["organization"] = organization
+        try:
+            result = self.configuration.call_remoteckan(
+                self.actions()["recommend_tags"], data
+            )
+        except Exception as e:
+            raise HDXError("Failed when trying to get recommended tags!") from e
+        return [tag["name"] for tag in result]
+
+    def is_allowed_for_datastore(self) -> bool:
+        """Return if the dataset or its organization is in HDX's allowlist for
+        automatic loading of resources into the datastore
+
+        Returns:
+            Whether dataset is allowed for datastore
+        """
+        success, result = self._read_from_hdx(
+            "dataset",
+            self.data["id"],
+            "package_id",
+            self.actions()["allowed_for_datastore"],
+        )
+        if success:
+            return result
+        logger.debug(result)
+        return False
+
+    def push_to_datastore(self) -> bool:
+        """Submit all the dataset's CSV resources to DataPusher+ using HDX's
+        wrapper action. Requires sysadmin.
+
+        Returns:
+            True if submitted, False if not (eg. no CSV resources)
+        """
+        result = self._write_to_hdx(
+            "push_to_datastore", {"dataset_id": self.data["id"]}, "dataset_id"
+        )
+        if not result["success"]:
+            logger.debug(result["message"])
+        return result["success"]
+
     def generate_resource(
         self,
         folder: Path | str,

src/hdx/data/location.py

@@ -43,6 +43,8 @@ def actions() -> dict[str, str]:
             "delete": "group_delete",
             "list": "group_list",
             "autocomplete": "group_autocomplete",
+            "data_grid": "hdx_datagrid_show",
+            "invalidate_data_grid": "invalidate_data_completeness_for_location",
         }
 
     @classmethod
@@ -218,6 +220,47 @@ def get_data_grid_countries(
             if len(group["name"]) == 3 and group.get("data_completeness") == "active"
         )
 
+    @staticmethod
+    def get_data_grid(
+        iso3: str, configuration: Configuration | None = None
+    ) -> dict | None:
+        """Get the Data Grid of a country. The dictionary has keys including
+        title, date, stats and categories. Each category has subcategories,
+        each with a list of datasets. Complementary datasets have
+        is_complementary set to True. Requires sysadmin.
+
+        Args:
+            iso3: Country ISO 3-letter code
+            configuration: HDX configuration. Defaults to global configuration.
+
+        Returns:
+            Data Grid dictionary or None if not found
+        """
+        location = Location(configuration=configuration)
+        success, result = location._read_from_hdx(
+            "location", iso3.lower(), "id", location.actions()["data_grid"]
+        )
+        if success:
+            return result
+        logger.debug(result)
+        return None
+
+    @staticmethod
+    def invalidate_data_grid(
+        iso3: str, configuration: Configuration | None = None
+    ) -> None:
+        """Clear HDX's cached Data Grid of a country. Requires sysadmin.
+
+        Args:
+            iso3: Country ISO 3-letter code
+            configuration: HDX configuration. Defaults to global configuration.
+
+        Returns:
+            None
+        """
+        location = Location(configuration=configuration)
+        location._write_to_hdx("invalidate_data_grid", {"name": iso3.lower()}, "name")
+
     def get_activities(self, all_pages: bool = False, **kwargs: Any) -> list[Activity]:
         """Get the activity stream of this location. One page of results (whose size
         is given by limit) is returned unless all_pages is True.