diff --git a/documentation/index.md b/documentation/index.md index c3aa1b4..c1be1d6 100755 --- a/documentation/index.md +++ b/documentation/index.md @@ -35,6 +35,7 @@ upload your datasets to HDX. - [Organization Management](#organization-management) - [Location Management](#location-management) - [Vocabulary Management](#vocabulary-management) + - [Activity Streams](#activity-streams) - [Pipeline State](#pipeline-state) - [Working Examples](#working-examples) - [Project Framework](#project-framework) @@ -1188,6 +1189,40 @@ If you want to add a list of tags, you do it as follows: vocabulary.add_tags(["TAG","TAG","TAG"...]) +## Activity Streams + +The **Activity** class wraps CKAN activities, which record who changed what and when. +Activities are read only. You can get the activity stream of a dataset, organization, +location or user: + + activities = dataset.get_activities(**kwargs) + activities = organization.get_activities(**kwargs) + activities = location.get_activities(**kwargs) + activities = user.get_activities(**kwargs) + +These return a list of **Activity** objects, each with fields like `timestamp`, +`user_id`, `object_id` and `activity_type` (eg. "changed package"). One page of results +is returned. Its size is set by `limit` (default 31, HDX maximum 100) and where it starts +by `offset`. Passing `all_pages=True` gets every page. Other arguments (`**kwargs`) like +`after`, `before` and `activity_types` depend on the object type and are detailed in the +API documentation. Getting a user's activity stream requires authorisation to view the +user's profile. + +You can get the activity stream of all recently added or changed datasets: + + activities = Activity.get_recently_changed_packages(limit=10) + +With `all_pages=True`, this stream covers the whole history of HDX so it is capped by +`max_activities` (default 1000). Pass `max_activities=None` to remove the cap. + +You can read a single activity, get its data (including a snapshot of the object at the +time of the activity) or get a diff against the object's previous version. HDX requires +permission to edit the object the activity relates to for these: + + activity = Activity.read_from_hdx("ACTIVITY_ID") + data = activity.get_data() + diff = activity.get_diff(diff_type="unified") # or "context" or "html" + ## Pipeline State The HDXState class allows the reading and writing of state to a given dataset. Input diff --git a/src/hdx/data/activity.py b/src/hdx/data/activity.py new file mode 100644 index 0000000..e07cdb4 --- /dev/null +++ b/src/hdx/data/activity.py @@ -0,0 +1,245 @@ +"""Activity class containing all logic for reading activity streams.""" + +import logging +from collections.abc import Sequence +from typing import Any, Optional + +from hdx.api.configuration import Configuration +from hdx.data.hdxobject import HDXError, HDXObject + +logger = logging.getLogger(__name__) + + +class Activity(HDXObject): + """Activity class containing all logic for reading activity streams. Activities + are read only. + + Args: + initial_data: Initial activity metadata dictionary. Defaults to None. + configuration: HDX configuration. Defaults to global configuration. + """ + + def __init__( + self, + initial_data: dict | None = None, + configuration: Configuration | None = None, + ) -> None: + if not initial_data: + initial_data = {} + super().__init__(initial_data, configuration=configuration) + + @staticmethod + def actions() -> dict[str, str]: + """Dictionary of actions that can be performed on object + + Returns: + Dictionary of actions that can be performed on object + """ + return { + "show": "activity_show", + "data_show": "activity_data_show", + "diff": "activity_diff", + "package_list": "package_activity_list", + "organization_list": "organization_activity_list", + "group_list": "group_activity_list", + "user_list": "user_activity_list", + "recently_changed_packages": "recently_changed_packages_activity_list", + } + + @classmethod + def read_from_hdx( + cls, identifier: str, configuration: Configuration | None = None + ) -> Optional["Activity"]: + """Reads the activity given by identifier from HDX and returns Activity object. + HDX requires permission to edit the object the activity relates to. + + Args: + identifier: Identifier of activity + configuration: HDX configuration. Defaults to global configuration. + + Returns: + Activity object if successful read, None if not + """ + return cls._read_from_hdx_class("activity", identifier, configuration) + + def get_object_type(self) -> str: + """Get the CKAN object type the activity relates to eg. package, + organization, group or user. This is the last word of the activity type + (eg. "changed package") with "dataset" (from "follow dataset") mapped to + "package". + + Returns: + Object type + """ + object_type = self.data["activity_type"].split(" ")[-1] + if object_type == "dataset": + return "package" + return object_type + + def get_data(self) -> dict: + """Get the data of the activity which includes a snapshot of the object + at the time of the activity. HDX requires permission to edit the object + the activity relates to. + + Returns: + Activity data + """ + return self._write_to_hdx("data_show", {"id": self.data["id"]}, "id") + + def get_diff(self, diff_type: str = "unified") -> dict: + """Get a diff of the object compared to its previous version. HDX requires + permission to edit the object the activity relates to. + + Args: + diff_type: One of unified, context or html. Defaults to unified. + + Returns: + Activity diff + """ + data = { + "id": self.data["id"], + "object_type": self.get_object_type(), + "diff_type": diff_type, + } + return self._write_to_hdx("diff", data, "id") + + @classmethod + def _get_list( + cls, + action: str, + data: dict, + configuration: Configuration | None = None, + all_pages: bool = False, + max_activities: int | None = None, + ) -> list["Activity"]: + """Helper method to get a list of activities from HDX + + Args: + action: Action to perform + data: Data to pass to CKAN including any offset and limit + configuration: HDX configuration. Defaults to global configuration. + all_pages: Whether to get all pages of results. Defaults to False. + max_activities: Maximum number of activities if all_pages is True. Defaults to None (no maximum). + + Returns: + List of Activity objects + """ + activity = cls(configuration=configuration) + if not all_pages: + result = activity._write_to_hdx(action, data) + return [cls(x, configuration=configuration) for x in result] + data = dict(data) + limit = data.setdefault("limit", 100) + offset = data.get("offset", 0) + activities = [] + while True: + data["offset"] = offset + result = activity._write_to_hdx(action, data) + activities.extend(cls(x, configuration=configuration) for x in result) + if max_activities is not None and len(activities) >= max_activities: + return activities[:max_activities] + if len(result) < limit: + return activities + offset += limit + + @classmethod + def get_activities( + cls, + object_type: str, + identifier: str, + configuration: Configuration | None = None, + all_pages: bool = False, + **kwargs: Any, + ) -> list["Activity"]: + """Get the activity stream of a CKAN object. One page of results (whose + size is given by limit) is returned unless all_pages is True. + + Args: + object_type: One of package, organization, group or user + identifier: Id or name of object + configuration: HDX configuration. Defaults to global configuration. + all_pages: Whether to get all pages of results. Defaults to False. + **kwargs: See below + offset (int): Where to start getting activity items from. Defaults to 0. + limit (int): Maximum number of activities per page. Defaults to 31 (HDX maximum 100) or 100 if all_pages is True. + after (int | str): Only activities after this timestamp (package and user only) + before (int | str): Only activities before this timestamp (package and user only) + include_hidden_activity (bool): Include hidden activity, sysadmins only (package, organization and group only) + activity_types (list): Activity types to include (package only) + exclude_activity_types (list): Activity types to exclude (package only) + + Returns: + List of Activity objects + """ + if not identifier: + raise HDXError(f"No id or name for {object_type} to get activities!") + data = {"id": identifier} + data.update(kwargs) + return cls._get_list(f"{object_type}_list", data, configuration, all_pages) + + @classmethod + def get_recently_changed_packages( + cls, + configuration: Configuration | None = None, + all_pages: bool = False, + max_activities: int | None = 1000, + **kwargs: Any, + ) -> list["Activity"]: + """Get the activity stream of all recently added or changed datasets. + One page of results (whose size is given by limit) is returned unless + all_pages is True. all_pages is capped by max_activities because the + stream covers the whole history of HDX. + + Args: + configuration: HDX configuration. Defaults to global configuration. + all_pages: Whether to get all pages of results. Defaults to False. + max_activities: Maximum number of activities if all_pages is True. Defaults to 1000. None means no maximum (very slow). + **kwargs: See below + offset (int): Where to start getting activity items from. Defaults to 0. + limit (int): Maximum number of activities per page. Defaults to 31 (HDX maximum 100) or 100 if all_pages is True. + + Returns: + List of Activity objects + """ + return cls._get_list( + "recently_changed_packages", + kwargs, + configuration, + all_pages, + max_activities, + ) + + def check_required_fields(self, ignore_fields: Sequence[str] = ()) -> None: + """Activities are read only so this raises an exception + + Args: + ignore_fields: Fields to ignore. Default is (). + + Returns: + None + """ + raise HDXError("Activities are read only!") + + def update_in_hdx(self, **kwargs: Any) -> None: + """Activities are read only so this raises an exception + + Returns: + None + """ + raise HDXError("Activities are read only!") + + def create_in_hdx(self, **kwargs: Any) -> None: + """Activities are read only so this raises an exception + + Returns: + None + """ + raise HDXError("Activities are read only!") + + def delete_from_hdx(self) -> None: + """Activities are read only so this raises an exception + + Returns: + None + """ + raise HDXError("Activities are read only!") diff --git a/src/hdx/data/dataset.py b/src/hdx/data/dataset.py index f8292bf..94f7fae 100755 --- a/src/hdx/data/dataset.py +++ b/src/hdx/data/dataset.py @@ -40,6 +40,7 @@ from hdx.api.locations import Locations from hdx.api.utilities.date_helper import DateHelper from hdx.api.utilities.filestore_helper import FilestoreHelper +from hdx.data.activity import Activity from hdx.data.hdxobject import HDXError, HDXObject from hdx.data.resource_matcher import ResourceMatcher @@ -1428,6 +1429,32 @@ def get_all_resources( resources.append(resource) return resources + def get_activities(self, all_pages: bool = False, **kwargs: Any) -> list[Activity]: + """Get the activity stream of this dataset. One page of results (whose size + is given by limit) is returned unless all_pages is True. + + Args: + all_pages: Whether to get all pages of results. Defaults to False. + **kwargs: See below + offset (int): Where to start getting activity items from. Defaults to 0. + limit (int): Maximum number of activities per page. Defaults to 31 (HDX maximum 100) or 100 if all_pages is True. + after (int | str): Only activities after this timestamp + before (int | str): Only activities before this timestamp + include_hidden_activity (bool): Include hidden activity (sysadmins only) + activity_types (list): Activity types to include + exclude_activity_types (list): Activity types to exclude + + Returns: + List of Activity objects + """ + return Activity.get_activities( + "package", + self.data.get("id") or self.data.get("name"), + self.configuration, + all_pages, + **kwargs, + ) + @classmethod def autocomplete( cls, diff --git a/src/hdx/data/location.py b/src/hdx/data/location.py index b2c791c..2a5367c 100644 --- a/src/hdx/data/location.py +++ b/src/hdx/data/location.py @@ -5,6 +5,7 @@ from typing import Any, Optional from hdx.api.configuration import Configuration +from hdx.data.activity import Activity from hdx.data.hdxobject import HDXObject logger = logging.getLogger(__name__) @@ -217,6 +218,28 @@ def get_data_grid_countries( if len(group["name"]) == 3 and group.get("data_completeness") == "active" ) + 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. + + Args: + all_pages: Whether to get all pages of results. Defaults to False. + **kwargs: See below + offset (int): Where to start getting activity items from. Defaults to 0. + limit (int): Maximum number of activities per page. Defaults to 31 (HDX maximum 100) or 100 if all_pages is True. + include_hidden_activity (bool): Include hidden activity (sysadmins only) + + Returns: + List of Activity objects + """ + return Activity.get_activities( + "group", + self.data.get("id") or self.data.get("name"), + self.configuration, + all_pages, + **kwargs, + ) + @classmethod def autocomplete( cls, diff --git a/src/hdx/data/organization.py b/src/hdx/data/organization.py index 4ccc39a..fd01bbe 100755 --- a/src/hdx/data/organization.py +++ b/src/hdx/data/organization.py @@ -8,6 +8,7 @@ import hdx.data.dataset import hdx.data.user as user_module from hdx.api.configuration import Configuration +from hdx.data.activity import Activity from hdx.data.hdxobject import HDXError, HDXObject if TYPE_CHECKING: @@ -311,6 +312,28 @@ def get_all_organization_names( return final_orgs + def get_activities(self, all_pages: bool = False, **kwargs: Any) -> list[Activity]: + """Get the activity stream of this organization. One page of results (whose size + is given by limit) is returned unless all_pages is True. + + Args: + all_pages: Whether to get all pages of results. Defaults to False. + **kwargs: See below + offset (int): Where to start getting activity items from. Defaults to 0. + limit (int): Maximum number of activities per page. Defaults to 31 (HDX maximum 100) or 100 if all_pages is True. + include_hidden_activity (bool): Include hidden activity (sysadmins only) + + Returns: + List of Activity objects + """ + return Activity.get_activities( + "organization", + self.data.get("id") or self.data.get("name"), + self.configuration, + all_pages, + **kwargs, + ) + @classmethod def autocomplete( cls, diff --git a/src/hdx/data/user.py b/src/hdx/data/user.py index 3d5c3e7..6af1e84 100755 --- a/src/hdx/data/user.py +++ b/src/hdx/data/user.py @@ -7,6 +7,7 @@ import hdx.data.organization from hdx.api.configuration import Configuration +from hdx.data.activity import Activity from hdx.data.hdxobject import HDXError, HDXObject logger = logging.getLogger(__name__) @@ -453,6 +454,30 @@ def get_token_list(self): return result return [] + def get_activities(self, all_pages: bool = False, **kwargs: Any) -> list[Activity]: + """Get the activity stream of this user. One page of results (whose size + is given by limit) is returned unless all_pages is True. HDX requires + authorisation to view the user's profile. + + Args: + all_pages: Whether to get all pages of results. Defaults to False. + **kwargs: See below + offset (int): Where to start getting activity items from. Defaults to 0. + limit (int): Maximum number of activities per page. Defaults to 31 (HDX maximum 100) or 100 if all_pages is True. + after (int | str): Only activities after this timestamp + before (int | str): Only activities before this timestamp + + Returns: + List of Activity objects + """ + return Activity.get_activities( + "user", + self.data.get("id") or self.data.get("name"), + self.configuration, + all_pages, + **kwargs, + ) + @classmethod def autocomplete( cls, diff --git a/tests/hdx/data/test_activity.py b/tests/hdx/data/test_activity.py new file mode 100644 index 0000000..f4dcfc7 --- /dev/null +++ b/tests/hdx/data/test_activity.py @@ -0,0 +1,196 @@ +"""Activity Tests""" + +import json + +import pytest + +from .. import MockResponse +from hdx.api.configuration import Configuration +from hdx.data.activity import Activity +from hdx.data.dataset import Dataset +from hdx.data.hdxobject import HDXError +from hdx.data.location import Location +from hdx.data.organization import Organization +from hdx.data.user import User + +activity_id = "7d18e25e-c004-4e9f-8f6b-a4b4fe8a0159" +activity = { + "id": activity_id, + "timestamp": "2026-10-14T01:49:59.740998", + "user_id": "52c8d2f3-a0d6-4cd9-8765-cf5cb4b67427", + "object_id": "38772da7-c0b0-477e-a60e-a385a9dd4d64", + "activity_type": "changed package", + "permission_labels": ["public"], +} +activity_data = {"package": {"name": "test-dataset"}, "actor": "testuser"} +activity_diff = {"diff": "--- a\n+++ b\n", "activities": []} + +total_activities = 250 +activities = [ + activity | {"id": f"activity{i}", "timestamp": str(i)} + for i in range(total_activities) +] + + +def mockresponse(result, action): + return MockResponse( + 200, + f'{{"success": true, "result": {json.dumps(result)}, "help": "http://test-data.humdata.org/api/3/action/help_show?name={action}"}}', + ) + + +def mocknotfound(action): + 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}"}}', + ) + + +class TestActivity: + calls = [] + + @pytest.fixture(scope="function") + def post_activity(self): + TestActivity.calls = [] + + class MockSession: + @staticmethod + def post(url, data, **kwargs): + datadict = json.loads(data.decode("utf-8")) + action = url.split("/")[-1] + TestActivity.calls.append((action, datadict)) + if action.endswith("_activity_list"): + if ( + action != "recently_changed_packages_activity_list" + and datadict["id"] != "TEST1" + ): + return mocknotfound(action) + offset = datadict.get("offset", 0) + limit = datadict.get("limit", 31) + return mockresponse(activities[offset : offset + limit], action) + if datadict["id"] != activity_id: + return mocknotfound(action) + if action == "activity_show": + return mockresponse(activity, action) + if action == "activity_data_show": + return mockresponse(activity_data, action) + if action == "activity_diff": + return mockresponse(activity_diff, action) + return mocknotfound(action) + + Configuration.read().remoteckan().session = MockSession() + + def test_read_from_hdx(self, configuration, post_activity): + result = Activity.read_from_hdx(activity_id) + assert result.data == activity + assert Activity.read_from_hdx("TEST2") is None + + def test_get_object_type(self, configuration): + assert Activity(activity).get_object_type() == "package" + result = Activity({"activity_type": "new organization"}).get_object_type() + assert result == "organization" + result = Activity({"activity_type": "follow dataset"}).get_object_type() + assert result == "package" + + def test_get_data_and_diff(self, configuration, post_activity): + result = Activity(activity) + assert result.get_data() == activity_data + assert result.get_diff() == activity_diff + assert TestActivity.calls[-1] == ( + "activity_diff", + {"id": activity_id, "object_type": "package", "diff_type": "unified"}, + ) + result.get_diff("html") + assert TestActivity.calls[-1][1]["diff_type"] == "html" + result = Activity({"id": "TEST2", "activity_type": "changed package"}) + with pytest.raises(HDXError): + result.get_data() + with pytest.raises(HDXError): + result.get_diff() + + def test_get_activities(self, configuration, post_activity): + result = Activity.get_activities("package", "TEST1") + assert len(result) == 31 + assert isinstance(result[0], Activity) + assert result[0]["id"] == "activity0" + result = Activity.get_activities("package", "TEST1", offset=10, limit=5) + assert [x["id"] for x in result] == [f"activity{i}" for i in range(10, 15)] + assert TestActivity.calls[-1] == ( + "package_activity_list", + {"id": "TEST1", "offset": 10, "limit": 5}, + ) + with pytest.raises(HDXError): + Activity.get_activities("package", "TEST2") + with pytest.raises(HDXError): + Activity.get_activities("package", None) + + def test_get_activities_all_pages(self, configuration, post_activity): + result = Activity.get_activities("package", "TEST1", all_pages=True) + assert [x["id"] for x in result] == [x["id"] for x in activities] + assert [x[1]["offset"] for x in TestActivity.calls] == [0, 100, 200] + TestActivity.calls = [] + result = Activity.get_activities( + "package", "TEST1", all_pages=True, offset=150, limit=50 + ) + assert [x["id"] for x in result] == [x["id"] for x in activities[150:]] + assert [x[1]["offset"] for x in TestActivity.calls] == [150, 200, 250] + + def test_get_recently_changed_packages(self, configuration, post_activity): + result = Activity.get_recently_changed_packages(limit=3) + assert [x["id"] for x in result] == ["activity0", "activity1", "activity2"] + assert TestActivity.calls[-1] == ( + "recently_changed_packages_activity_list", + {"limit": 3}, + ) + TestActivity.calls = [] + result = Activity.get_recently_changed_packages(all_pages=True) + assert len(result) == total_activities + assert [x[1]["offset"] for x in TestActivity.calls] == [0, 100, 200] + TestActivity.calls = [] + result = Activity.get_recently_changed_packages( + all_pages=True, max_activities=120 + ) + assert [x["id"] for x in result] == [x["id"] for x in activities[:120]] + assert [x[1]["offset"] for x in TestActivity.calls] == [0, 100] + result = Activity.get_recently_changed_packages( + all_pages=True, max_activities=100 + ) + assert len(result) == 100 + result = Activity.get_recently_changed_packages( + all_pages=True, max_activities=None, limit=50 + ) + assert len(result) == total_activities + + @pytest.mark.parametrize( + "hdxobjectclass,action", + [ + (Dataset, "package_activity_list"), + (Organization, "organization_activity_list"), + (Location, "group_activity_list"), + (User, "user_activity_list"), + ], + ) + def test_hdxobject_get_activities( + self, configuration, post_activity, hdxobjectclass, action + ): + hdxobject = hdxobjectclass({"name": "TEST1"}) + result = hdxobject.get_activities(limit=2) + assert [x["id"] for x in result] == ["activity0", "activity1"] + assert isinstance(result[0], Activity) + assert TestActivity.calls[-1] == (action, {"id": "TEST1", "limit": 2}) + hdxobject = hdxobjectclass({"id": "TEST1", "name": "TEST2"}) + hdxobject.get_activities() + assert TestActivity.calls[-1] == (action, {"id": "TEST1"}) + with pytest.raises(HDXError): + hdxobjectclass().get_activities() + + def test_read_only(self, configuration): + result = Activity(activity) + with pytest.raises(HDXError): + result.check_required_fields() + with pytest.raises(HDXError): + result.create_in_hdx() + with pytest.raises(HDXError): + result.update_in_hdx() + with pytest.raises(HDXError): + result.delete_from_hdx()