Add data dictionary methods

documentation/index.md

@@ -937,6 +937,33 @@ 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.
 
+### Data Dictionary
+
+A resource can have a data dictionary describing the columns in the data it
+contains. It is a list of dictionaries, each with the keys `field` (the
+column name as it appears in the file), `label` (a human-readable label) and
+`description` (a human-readable description of the column). Set it using
+**set_hdx_data_dictionary**:
+
+    resource.set_hdx_data_dictionary([
+        {
+            "field": "col_name",
+            "label": "Column Label",
+            "description": "Human-readable description.",
+        },
+        {
+            "field": "another_col",
+            "label": "Another Label",
+            "description": "Second column description.",
+        },
+    ])
+
+`field`, `label` and `description` must all be non-empty strings for every
+column, otherwise an `HDXError` is raised. Read the data dictionary back
+using the getter:
+
+    data_dictionary = resource.get_hdx_data_dictionary()
+
 ## Showcase Management
 
 The **Showcase** class enables you to manage showcases, creating, deleting and updating

pyproject.toml

@@ -39,8 +39,8 @@ dependencies = [
     "ckanapi>=4.11",
     "defopt>=7.0.0",
     "email_validator",
-    "hdx-python-country>=4.1.1",
-    "hdx-python-utilities>=4.0.8",
+    "hdx-python-country>=4.1.3",
+    "hdx-python-utilities>=4.1.2",
     "makefun",
     "requests",
 ]

src/hdx/data/resource.py

@@ -1,5 +1,6 @@
 """Resource class containing all logic for creating, checking, and updating resources."""
 
+import json
 import logging
 from collections.abc import Sequence
 from datetime import datetime
@@ -33,6 +34,9 @@ class Resource(HDXObject):
     """
 
     _formats_dict = None
+    _valid_data_dictionary_types = frozenset(
+        {"text", "numeric", "date", "timestamp without time zone"}
+    )
 
     def __init__(
         self,
@@ -263,6 +267,56 @@ def set_format(self, format: str) -> str:
         self.data["format"] = file_format
         return file_format
 
+    def get_hdx_data_dictionary(self) -> list[dict] | None:
+        """Get the resource's data dictionary (column definitions)
+
+        Returns:
+            List of column definition dictionaries or None if not set
+        """
+        data_dictionary = self.data.get("hdx_data_dictionary")
+        if data_dictionary is None:
+            return None
+        if isinstance(data_dictionary, str):
+            return json.loads(data_dictionary)
+        return data_dictionary
+
+    def set_hdx_data_dictionary(self, data_dictionary: Sequence[dict]) -> None:
+        """Set the resource's data dictionary (column definitions). Each column
+        definition must be a dict with non-empty string values for field
+        (column name in the CSV file), label (human-readable column label) and
+        description (human-readable description of the column). If present,
+        data_type must be one of the PostgreSQL types produced by DataPusher+
+        (text, numeric, date, timestamp without time zone).
+
+        Args:
+            data_dictionary: Sequence of column definition dictionaries
+
+        Returns:
+            None
+        """
+        if not data_dictionary:
+            raise HDXError("hdx_data_dictionary must be a non-empty list!")
+        for i, column in enumerate(data_dictionary):
+            if not isinstance(column, dict):
+                raise HDXError(f"hdx_data_dictionary[{i}] must be a dict!")
+            for required_key in ("field", "label", "description"):
+                value = column.get(required_key)
+                if not isinstance(value, str) or not value.strip():
+                    raise HDXError(
+                        f"hdx_data_dictionary[{i}] is missing a non-empty '{required_key}'!"
+                    )
+            if "data_type" in column:
+                data_type = column["data_type"]
+                if data_type not in self._valid_data_dictionary_types:
+                    valid_types = ", ".join(sorted(self._valid_data_dictionary_types))
+                    raise HDXError(
+                        f"hdx_data_dictionary[{i}] has invalid 'data_type' "
+                        f"'{data_type}'! Must be one of: {valid_types}"
+                    )
+        self.data["hdx_data_dictionary"] = json.dumps(
+            data_dictionary, separators=(",", ":")
+        )
+
     def clean_format(self) -> str:
         """Clean the resource's format, setting it to None if it is invalid and
         cannot be mapped

tests/hdx/data/test_resource.py

@@ -734,6 +734,70 @@ def test_get_set_date_of_resource(self, configuration):
             "ongoing": False,
         }
 
+    def test_get_set_hdx_data_dictionary(self, configuration):
+        resource = Resource({})
+        assert resource.get_hdx_data_dictionary() is None
+
+        data_dictionary = [
+            {"field": "col_name", "label": "Column Label", "description": "Desc."},
+            {
+                "field": "another_col",
+                "label": "Another Label",
+                "description": "Second column description.",
+            },
+        ]
+        resource.set_hdx_data_dictionary(data_dictionary)
+        assert resource.data["hdx_data_dictionary"] == json.dumps(
+            data_dictionary, separators=(",", ":")
+        )
+        assert resource.get_hdx_data_dictionary() == data_dictionary
+
+        resource2 = Resource({"hdx_data_dictionary": json.dumps(data_dictionary)})
+        assert resource2.get_hdx_data_dictionary() == data_dictionary
+
+        with pytest.raises(HDXError):
+            resource.set_hdx_data_dictionary([])
+        with pytest.raises(HDXError):
+            resource.set_hdx_data_dictionary([{"field": "col_name", "label": "Label"}])
+        with pytest.raises(HDXError):
+            resource.set_hdx_data_dictionary(
+                [{"field": "", "label": "Label", "description": "Desc."}]
+            )
+        with pytest.raises(HDXError):
+            resource.set_hdx_data_dictionary(
+                [{"field": "col_name", "label": "  ", "description": "Desc."}]
+            )
+        with pytest.raises(HDXError):
+            resource.set_hdx_data_dictionary(["not a dict"])
+        with pytest.raises(HDXError):
+            resource.set_hdx_data_dictionary(
+                [
+                    {
+                        "field": "col_name",
+                        "label": "Label",
+                        "description": "Desc.",
+                        "data_type": "varchar",
+                    }
+                ]
+            )
+
+        data_dictionary_with_types = [
+            {
+                "field": "col_name",
+                "label": "Column Label",
+                "description": "Desc.",
+                "data_type": "numeric",
+            },
+            {
+                "field": "another_col",
+                "label": "Another Label",
+                "description": "Second column description.",
+                "data_type": "timestamp without time zone",
+            },
+        ]
+        resource.set_hdx_data_dictionary(data_dictionary_with_types)
+        assert resource.get_hdx_data_dictionary() == data_dictionary_with_types
+
     def test_check_required_fields(self, configuration):
         resource_data_copy = copy.deepcopy(resource_data)
         resource = Resource(resource_data_copy)