Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 35 additions & 0 deletions documentation/index.md
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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
Expand Down
245 changes: 245 additions & 0 deletions src/hdx/data/activity.py
Original file line number Diff line number Diff line change
@@ -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!")
27 changes: 27 additions & 0 deletions src/hdx/data/dataset.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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,
Expand Down
23 changes: 23 additions & 0 deletions src/hdx/data/location.py
Original file line number Diff line number Diff line change
Expand Up @@ -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__)
Expand Down Expand Up @@ -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,
Expand Down
Loading
Loading