HDXDSYS-3013 Add HDX custom extensions like fs_check_info_show

documentation/index.md

@@ -36,6 +36,7 @@ upload your datasets to HDX.
     -   [Location Management](#location-management)
     -   [Vocabulary Management](#vocabulary-management)
     -   [Activity Streams](#activity-streams)
+    -   [HDX Statistics](#hdx-statistics)
     -   [Pipeline State](#pipeline-state)
 -   [Working Examples](#working-examples)
 -   [Project Framework](#project-framework)
@@ -636,6 +637,16 @@ function, for example:
 
     dataset.remove_showcase(showcase)
 
+HDX runs file structure checks on resources, recording schema changes (eg.
+column names, order or types) and whether sheets have merged cells. You can get
+these checks for all of a dataset's resources from HDX:
+
+    fs_check_info = dataset.get_fs_check_info()
+
+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`.
+
 ### Time Period
 
 Time Period is a mandatory field in HDX. It is the earliest start date and
@@ -947,6 +958,12 @@ If the method **set_file_to_upload** is used to supply a file, the resource
 `last_modified` field is set to now automatically regardless of the value of
 `data_updated` or whether **mark_data_updated** has been called.
 
+The file structure check events of a resource (see Dataset Specific
+Operations) can be obtained from its `fs_check_info` field without a call to
+HDX:
+
+    events = resource.get_fs_check_info()
+
 ### Data Dictionary
 
 A resource can have a data dictionary describing the columns in the data it
@@ -1074,6 +1091,11 @@ configuration):
 
     user = User.get_current_user()
 
+You can get the name of the API token in the configuration and the md5 hash
+of its owner's email:
+
+    token_info = User.get_current_token_info()  # {"email_hash": ..., "token_name": ...}
+
 You can check that the current user has a particular permission to a specific
 organization:
 
@@ -1230,6 +1252,20 @@ permission to edit the object the activity relates to for these:
     data = activity.get_data()
     diff = activity.get_diff(diff_type="unified")  # or "context" or "html"
 
+## HDX Statistics
+
+General statistics about datasets, organizations and users on HDX can be
+obtained as follows:
+
+    from hdx.api.utilities.hdx_stats import get_general_statistics, push_general_stats
+
+    stats = get_general_statistics()
+
+Users with permission to manage basic scheduled tasks can compute the statistics
+and push them to analytics:
+
+    pushed_stats = push_general_stats()
+
 ## Pipeline State
 
 The HDXState class allows the reading and writing of state to a given dataset. Input

src/hdx/api/utilities/hdx_stats.py

@@ -0,0 +1,47 @@
+"""HDX site-wide statistics"""
+
+from hdx.api.configuration import Configuration
+from hdx.data.hdxobject import HDXError
+
+
+def _call_stats_action(action: str, configuration: Configuration | None) -> dict:
+    """Call an HDX statistics action
+
+    Args:
+        action: CKAN action name
+        configuration: HDX configuration. Defaults to global configuration.
+
+    Returns:
+        Statistics dictionary
+    """
+    if configuration is None:
+        configuration = Configuration.read()
+    try:
+        return configuration.call_remoteckan(action, {})
+    except Exception as e:
+        raise HDXError(f"Failed when trying to call {action}!") from e
+
+
+def get_general_statistics(configuration: Configuration | None = None) -> dict:
+    """Get general statistics about datasets, organizations and users on HDX
+
+    Args:
+        configuration: HDX configuration. Defaults to global configuration.
+
+    Returns:
+        Dictionary with keys datasets, organizations and users
+    """
+    return _call_stats_action("hdx_general_statistics", configuration)
+
+
+def push_general_stats(configuration: Configuration | None = None) -> dict:
+    """Compute general statistics on HDX and push them to analytics. Requires
+    permission to manage basic scheduled tasks.
+
+    Args:
+        configuration: HDX configuration. Defaults to global configuration.
+
+    Returns:
+        Statistics that were pushed
+    """
+    return _call_stats_action("hdx_push_general_stats", configuration)

src/hdx/data/dataset.py

@@ -147,6 +147,7 @@ def actions() -> dict[str, str]:
             "list": "package_list",
             "autocomplete": "package_autocomplete",
             "create_default_views": "package_create_default_resource_views",
+            "fs_check_info": "fs_check_info_show",
         }
 
     def __setitem__(self, key: Any, value: Any) -> None:
@@ -2333,6 +2334,25 @@ def get_api_url(self, prefer_name: bool = True) -> str | None:
             return None
         return f"{self.configuration.get_hdx_site_url()}/api/3/action/package_show?id={name_or_id}"
 
+    def get_fs_check_info(self) -> list[dict]:
+        """Get file structure check information for the dataset's resources
+        from HDX. Each list item is a dictionary from resource name to a list
+        of check events, each of which can have keys such as state, timestamp,
+        sheet_changes and hxl_proxy_response.
+
+        Returns:
+            List of dictionaries from resource name to file structure checks
+        """
+        success, result = self._read_from_hdx(
+            "dataset",
+            self.get_name_or_id(prefer_name=False),
+            "id",
+            self.actions()["fs_check_info"],
+        )
+        if success:
+            return result
+        return []
+
     def generate_resource(
         self,
         folder: Path | str,

src/hdx/data/resource.py

@@ -1174,6 +1174,21 @@ def mark_broken(self) -> None:
         else:
             logger.debug(result)
 
+    def get_fs_check_info(self) -> list[dict]:
+        """Get file structure check information for the resource from its
+        fs_check_info field. Each check event can have keys such as state,
+        timestamp, sheet_changes and hxl_proxy_response.
+
+        Returns:
+            List of file structure check events
+        """
+        fs_check_info = self.data.get("fs_check_info")
+        if not fs_check_info:
+            return []
+        if isinstance(fs_check_info, str):
+            return json.loads(fs_check_info)
+        return fs_check_info
+
     def is_marked_data_updated(self) -> bool:
         """Return if the resource's data is marked to be updated
 

src/hdx/data/user.py

@@ -45,6 +45,7 @@ def actions() -> dict[str, str]:
             "list": "user_list",
             "listorgs": "organization_list_for_user",
             "token_list": "api_token_list",
+            "token_info": "hdx_token_info",
             "autocomplete": "user_autocomplete",
         }
 
@@ -478,6 +479,24 @@ def get_activities(self, all_pages: bool = False, **kwargs: Any) -> list[Activit
             **kwargs,
         )
 
+    @classmethod
+    def get_current_token_info(cls, configuration: Configuration | None = None) -> dict:
+        """Get information about the API token in the configuration: the md5
+        hash of the token owner's email (email_hash) and the token name
+        (token_name).
+
+        Args:
+            configuration: HDX configuration. Defaults to global configuration.
+
+        Returns:
+            Dictionary with keys email_hash and token_name
+        """
+        user = User(configuration=configuration)
+        try:
+            return user.configuration.call_remoteckan(cls.actions()["token_info"], {})
+        except Exception as e:
+            raise HDXError("Failed when trying to get API token info!") from e
+
     @classmethod
     def autocomplete(
         cls,

tests/hdx/api/utilities/test_hdx_stats.py

@@ -0,0 +1,59 @@
+"""HDX Stats Tests"""
+
+import json
+
+import pytest
+
+from ... import MockResponse
+from hdx.api.configuration import Configuration
+from hdx.api.utilities.hdx_stats import get_general_statistics, push_general_stats
+from hdx.data.hdxobject import HDXError
+
+general_statistics = {
+    "datasets": {"total": 100, "with_geodata": 20, "with_showcases": 10},
+    "organizations": {"total": 30, "with_datasets": 25},
+    "users": {"total": 1000},
+}
+
+pushed_stats = {"mixpanel_meta": {"datasets total": 100, "orgs total": 30}}
+
+
+class TestHDXStats:
+    succeed = True
+
+    @pytest.fixture(scope="function")
+    def post_stats(self):
+        class MockSession:
+            @staticmethod
+            def post(url, data, **kwargs):
+                if TestHDXStats.succeed:
+                    if "hdx_general_statistics" in url:
+                        result = json.dumps(general_statistics)
+                    elif "hdx_push_general_stats" in url:
+                        result = json.dumps(pushed_stats)
+                    else:
+                        result = None
+                    if result:
+                        return MockResponse(
+                            200,
+                            f'{{"success": true, "result": {result}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=stats"}}',
+                        )
+                return MockResponse(
+                    403,
+                    '{"success": false, "error": {"message": "Access denied", "__type": "Authorization Error"}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=stats"}',
+                )
+
+        Configuration.read().remoteckan().session = MockSession()
+
+    def test_get_general_statistics(self, configuration, post_stats):
+        assert get_general_statistics() == general_statistics
+        assert get_general_statistics(Configuration.read()) == general_statistics
+
+    def test_push_general_stats(self, configuration, post_stats):
+        assert push_general_stats() == pushed_stats
+        TestHDXStats.succeed = False
+        with pytest.raises(HDXError):
+            push_general_stats()
+        with pytest.raises(HDXError):
+            get_general_statistics()
+        TestHDXStats.succeed = True

tests/hdx/data/test_dataset_noncore.py

@@ -37,6 +37,19 @@
 from hdx.data.user import User
 from hdx.data.vocabulary import Vocabulary
 
+fs_check_info = [
+    {
+        "MyResource1": [
+            {
+                "state": "success",
+                "timestamp": "2026-10-01T10:00:00",
+                "sheet_changes": {"nrows": 10},
+                "hxl_proxy_response": {"has_merged_cells": False},
+            }
+        ]
+    }
+]
+
 
 class TestDatasetNoncore:
     association = None
@@ -139,6 +152,31 @@ def post(url, data, **kwargs):
 
         Configuration.read().remoteckan().session = MockSession()
 
+    @pytest.fixture(scope="function")
+    def post_fs_check_info(self):
+        class MockSession:
+            @staticmethod
+            def post(url, data, **kwargs):
+                datadict = json.loads(data.decode("utf-8"))
+                if "fs_check_info_show" in url and datadict["id"] == "TEST1":
+                    result = json.dumps(fs_check_info)
+                    return MockResponse(
+                        200,
+                        f'{{"success": true, "result": {result}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=fs_check_info_show"}}',
+                    )
+                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=fs_check_info_show"}',
+                )
+
+        Configuration.read().remoteckan().session = MockSession()
+
+    def test_get_fs_check_info(self, configuration, post_fs_check_info):
+        dataset = Dataset({"id": "TEST1", "name": "test-dataset"})
+        assert dataset.get_fs_check_info() == fs_check_info
+        dataset = Dataset({"id": "TEST2"})
+        assert dataset.get_fs_check_info() == []
+
     def test_get_name_or_id(self, configuration, hdx_config_yaml, project_config_yaml):
         dataset = Dataset()
         assert dataset.get_name_or_id() is None

tests/hdx/data/test_resource.py

@@ -1327,6 +1327,22 @@ def test_date_data_updated(self, configuration, read):
             == expected
         )
 
+    def test_get_fs_check_info(self, configuration):
+        resource = Resource({"name": "MyResource1"})
+        assert resource.get_fs_check_info() == []
+        events = [
+            {
+                "state": "success",
+                "timestamp": "2026-10-01T10:00:00",
+                "sheet_changes": {"nrows": 10},
+                "hxl_proxy_response": {"has_merged_cells": False},
+            }
+        ]
+        resource["fs_check_info"] = json.dumps(events)
+        assert resource.get_fs_check_info() == events
+        resource["fs_check_info"] = events
+        assert resource.get_fs_check_info() == events
+
     def test_get_hdx_url(self, configuration, read):
         resource = Resource.read_from_hdx("74b74ae1-df0c-4716-829f-4f939a046811")
         assert (

tests/hdx/data/test_user.py

@@ -105,6 +105,7 @@ def mocklist(url):
 
 
 class TestUser:
+    token_valid = True
     smtp_initargs = {
         "host": "localhost",
         "port": 123,
@@ -398,6 +399,23 @@ def post(url, data, **kwargs):
 
         Configuration.read().remoteckan().session = MockSession()
 
+    @pytest.fixture(scope="function")
+    def post_tokeninfo(self):
+        class MockSession:
+            @staticmethod
+            def post(url, data, **kwargs):
+                if "hdx_token_info" in url and TestUser.token_valid:
+                    return MockResponse(
+                        200,
+                        '{"success": true, "result": {"email_hash": "9e2fc15d8cc65f43136f9dcef05fa184", "token_name": "my_token"}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=hdx_token_info"}',
+                    )
+                return MockResponse(
+                    409,
+                    '{"success": false, "error": {"message": "Invalid token", "__type": "Validation Error"}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=hdx_token_info"}',
+                )
+
+        Configuration.read().remoteckan().session = MockSession()
+
     @pytest.fixture(scope="function")
     def post_autocomplete(self):
         class MockSession:
@@ -728,5 +746,15 @@ def test_get_token_list(self, configuration, post_tokenlist):
         user["name"] = "lala"
         assert user.get_token_list() == []
 
+    def test_get_current_token_info(self, configuration, post_tokeninfo):
+        assert User.get_current_token_info() == {
+            "email_hash": "9e2fc15d8cc65f43136f9dcef05fa184",
+            "token_name": "my_token",
+        }
+        TestUser.token_valid = False
+        with pytest.raises(HDXError):
+            User.get_current_token_info()
+        TestUser.token_valid = True
+
     def test_autocomplete(self, configuration, post_autocomplete):
         assert User.autocomplete("fake%20test") == user_autocomplete