Fix activities

Author: vatj

python/hopsworks_common/core/feature_store_activity_api.py

@@ -41,16 +41,26 @@ def get_feature_group_activities(
             feature_store_id,
             "featuregroups",
             feature_group_id,
-            "activities",
+            "activity",
         ]
 
         query_params = {
             "limit": limit,
             "offset": offset,
+            "expand": [
+                "users",
+                "commits",
+                "jobs",
+                "validationreport",
+                "expectationsuite",
+                "executions",
+                "statistics",
+            ],
+            "sortby": "timestamp:desc",
         }
         if activity_type is not None and len(activity_type) > 0:
             query_params["activityType"] = activity_type
 
-        response = _client._request("GET", path_params, query_params=query_params)
+        response = _client._send_request("GET", path_params, query_params=query_params)
 
         return fsa_mod.FeatureStoreActivity.from_response_json(response)

python/hopsworks_common/feature_store_activity.py

@@ -21,8 +21,12 @@
 from enum import Enum
 from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union
 
+import humps
+from hopsworks import execution as execution_mod
 from hopsworks_common import user as user_mod
 from hopsworks_common import util
+from hsfs import expectation_suite as es_mod
+from hsfs import validation_report as vr_mod
 from hsfs.core.constants import HAS_GREAT_EXPECTATIONS
 
 
@@ -31,8 +35,6 @@
         ExpectationSuite,
         ExpectationSuiteValidationResult,
     )
-    from hsfs import expectation_suite as es_mod
-    from hsfs import validation_report as vr_mod
 
 
 class FeatureStoreActivityType(Enum):
@@ -44,69 +46,67 @@ class FeatureStoreActivityType(Enum):
     COMMIT = "COMMIT"
 
 
-@dataclass(frozen=True, init=False, repr=False)
+@dataclass(init=False, repr=False)
 class FeatureStoreActivity:
     type: FeatureStoreActivityType
-    metadata: str
     timestamp: int
-    user: user_mod.User
+    metadata: str
+    user: Optional[user_mod.User]
     # optional fields depending on the activity type
     validation_report: Optional[
         Union[vr_mod.ValidationReport, ExpectationSuiteValidationResult]
     ] = None
     expectation_suite: Optional[Union[es_mod.ExpectationSuite, ExpectationSuite]] = None
     commit: Optional[Dict[str, Union[str, int, float]]] = None
     statistics: Optional[Dict[str, Union[str, int, float]]] = None
+    execution: Optional[execution_mod.Execution] = None
+    execution_last_event_time: Optional[int] = None
     # internal fields
     id: int
     href: str
 
     def __init__(
         self,
-        type: FeatureStoreActivityType,
-        metadata: str,
+        type: str,
         timestamp: int,
-        user: Dict[str, Any],
+        metadata: Optional[str] = None,
+        user: Optional[Dict[str, Any]] = None,
+        expectation_suite: Optional[Dict[str, Any]] = None,
+        validation_report: Optional[Dict[str, Any]] = None,
+        commit: Optional[Dict[str, Union[str, int, float]]] = None,
+        statistics: Optional[Dict[str, Union[str, int, float]]] = None,
+        execution: Optional[Dict[str, Any]] = None,
+        execution_last_event_time: Optional[int] = None,
         **kwargs,
     ):
-        self.type = type
-        self.metadata = metadata
+        self.type = FeatureStoreActivityType(type) if isinstance(type, str) else type
         self.timestamp = timestamp
-        self.user = user_mod.User.from_response_json(user)
 
         self.id = kwargs.get("id")
-        self.href = kwargs.get("href", "")
+        self.href = kwargs.get("href")
 
-        self.commit = None
-        self.expectation_suite = None
-        self.validation_report = None
-        self.statistics = None
+        self.user = user_mod.User.from_response_json(user) if user else None
+        self.metadata = metadata
+        self.commit = commit
+        self.statistics = statistics
+        self.execution = (
+            execution_mod.Execution.from_response_json(execution) if execution else None
+        )
+        self.execution_last_event_time = execution_last_event_time
 
-        if self.type == FeatureStoreActivityType.VALIDATIONS:
-            if HAS_GREAT_EXPECTATIONS:
-                self.validation_report = ExpectationSuiteValidationResult(
-                    **kwargs.get("validation_report")
-                )
-            else:
-                self.validation_report = vr_mod.ValidationReport(
-                    **kwargs.get("validation_report")
-                )
-
-        if self.type == FeatureStoreActivityType.EXPECTATIONS:
+        if self.type == FeatureStoreActivityType.VALIDATIONS and validation_report:
+            self.validation_report = vr_mod.ValidationReport.from_response_json(
+                validation_report
+            )
             if HAS_GREAT_EXPECTATIONS:
-                self.expectation_suite = ExpectationSuite(
-                    **kwargs.get("expectation_suite")
-                )
-            else:
-                self.expectation_suite = es_mod.ExpectationSuite(
-                    **kwargs.get("expectation_suite")
-                )
-
-        if self.type == FeatureStoreActivityType.COMMIT:
-            self.commit = kwargs.get("commit")
+                self.validation_report = self.validation_report.to_ge_type()
 
-        if self.statistics:
-            self.statistics = kwargs.get("statistics")
+        if self.type == FeatureStoreActivityType.EXPECTATIONS and expectation_suite:
+            self.expectation_suite = es_mod.ExpectationSuite.from_response_json(
+                expectation_suite
+            )
+            if HAS_GREAT_EXPECTATIONS:
+                self.expectation_suite = self.expectation_suite.to_ge_type()
 
     @classmethod
     def from_response_json(
@@ -117,35 +117,67 @@ def from_response_json(
                 cls.from_response_json(activity) for activity in response_json["items"]
             ]
         else:
-            return cls(**response_json)
+            return cls(**humps.decamelize(response_json))
 
     def to_dict(self) -> Dict[str, Any]:
-        json = {
+        activity_dict = {
             "id": self.id,
             "type": self.type.value,
             "metadata": self.metadata,
             "timestamp": self.timestamp,
-            "user": self.user.json(),
         }
+        if self.user:
+            activity_dict["user"] = self.user.to_dict()
         if self.validation_report:
-            json["validation_report"] = self.validation_report.json()
+            activity_dict["validation_report"] = (
+                self.validation_report.to_dict()
+                if hasattr(self.validation_report, "_id")
+                else self.validation_report.to_json_dict()
+            )
         if self.expectation_suite:
-            json["expectation_suite"] = self.expectation_suite.json()
+            activity_dict["expectation_suite"] = (
+                self.expectation_suite.to_dict()
+                if hasattr(self.expectation_suite, "_id")
+                else self.expectation_suite.to_json_dict()
+            )
         if self.commit:
-            json["commit"] = self.commit
+            activity_dict["commit"] = self.commit
         if self.statistics:
-            json["statistics"] = self.statistics
+            activity_dict["statistics"] = self.statistics
+        if self.execution:
+            activity_dict["execution"] = humps.decamelize(
+                json.loads(self.execution.json())
+            )
 
-        return json
+        return activity_dict
 
     def json(self) -> str:
         return json.dumps(self.to_dict(), cls=util.Encoder)
 
     def __repr__(self):
         utc_human_readable = (
             datetime.datetime.fromtimestamp(
-                self.timestamp, datetime.timezone.utc
+                self.timestamp / 1000, datetime.timezone.utc
             ).strftime(r"%Y-%m-%d %H:%M:%S")
             + " UTC"
         )
-        return f"Activity:\n{self.type},\n{self.metadata},\nat: {utc_human_readable},\nuser: {self.user}"
+        the_string = f"Activity {self.type.value},"
+        the_string += f" at: {utc_human_readable}"
+        if self.user:
+            the_string += f", by: {self.user.email}"
+        if self.metadata:
+            the_string += f"\n\t{self.metadata}"
+        if self.execution:
+            the_string += f"\n{self.execution.get_url()},"
+        if self.validation_report:
+            the_string += f"Validation {'succeeded' if self.validation_report.success else 'failed'}."
+        if self.expectation_suite:
+            the_string += (
+                f"It has {len(self.expectation_suite.expectations)} expectations."
+            )
+        if self.statistics:
+            the_string += f"\nComputed following statistics:\n{json.dumps(self.statistics, indent=2)}"
+        if self.commit:
+            the_string += f"\nData ingestion:\n{json.dumps(self.commit, indent=2)}"
+
+        return the_string

python/hopsworks_common/user.py

@@ -17,122 +17,51 @@
 from __future__ import annotations
 
 import json
+from dataclasses import asdict, dataclass
+from typing import Any, Dict, Optional
 
 import humps
 from hopsworks_common import util
 
 
+@dataclass
 class User:
-    def __init__(
-        self,
-        username=None,
-        email=None,
-        first_name=None,
-        last_name=None,
-        status=None,
-        secret=None,
-        chosen_password=None,
-        repeated_password=None,
-        tos=None,
-        two_factor=None,
-        tours_state=None,
-        max_num_projects=None,
-        test_user=None,
-        user_account_type=None,
-        num_active_projects=None,
-        num_remaining_projects=None,
-        firstname=None,
-        lastname=None,
-        **kwargs,
-    ):
-        self._username = username
-        self._email = email
-        self._first_name = first_name or firstname
-        self._last_name = last_name or last_name
-        self._status = status
-        self._secret = secret
-        self._chosen_password = chosen_password
-        self._repeated_password = repeated_password
-        self._tos = tos
-        self._two_factor = two_factor
-        self._tours_state = tours_state
-        self._max_num_projects = max_num_projects
-        self._test_user = test_user
-        self._user_account_type = user_account_type
-        self._num_active_projects = num_active_projects
-        self._num_remaining_projects = num_remaining_projects
+    email: str
+    first_name: str
+    last_name: str
+    href: Optional[str] = None
+    username: Optional[str] = None
+    status: Optional[str] = None
+    secret: Optional[str] = None
+    chosen_password: Optional[str] = None
+    repeated_password: Optional[str] = None
+    tos: Optional[bool] = None
+    two_factor: Optional[bool] = None
+    user_account_type: Optional[str] = None
+    tours_state: Optional[bool] = None
+    max_num_projects: Optional[int] = None
+    test_user: Optional[bool] = None
+    num_active_projects: Optional[int] = None
+    num_remaining_projects: Optional[int] = None
 
     @classmethod
-    def from_response_json(cls, json_dict):
+    def from_response_json(cls, json_dict: Optional[Dict[str, Any]]) -> Optional[User]:
         if json_dict:
             json_decamelized = humps.decamelize(json_dict)
+            if "firstname" in json_decamelized.keys():
+                json_decamelized["first_name"] = json_decamelized.pop("firstname")
+                json_decamelized["last_name"] = json_decamelized.pop("lastname")
+            # Remove keys that are not part of the dataclass
+            for key in set(json_decamelized.keys()) - set(
+                User.__dataclass_fields__.keys()
+            ):
+                json_decamelized.pop(key)
             return cls(**json_decamelized)
         else:
             return None
 
-    def json(self):
+    def json(self) -> str:
         return json.dumps(self, cls=util.Encoder)
 
-    @property
-    def username(self):
-        return self._username
-
-    @property
-    def email(self):
-        return self._email
-
-    @property
-    def first_name(self):
-        return self._first_name
-
-    @property
-    def last_name(self):
-        return self._last_name
-
-    @property
-    def status(self):
-        return self._status
-
-    @property
-    def secret(self):
-        return self._secret
-
-    @property
-    def chosen_password(self):
-        return self._chosen_password
-
-    @property
-    def repeated_password(self):
-        return self._repeated_password
-
-    @property
-    def tos(self):
-        return self._tos
-
-    @property
-    def two_factor(self):
-        return self._two_factor
-
-    @property
-    def tours_state(self):
-        return self._tours_state
-
-    @property
-    def max_num_projects(self):
-        return self._max_num_projects
-
-    @property
-    def test_user(self):
-        return self._test_user
-
-    @property
-    def user_account_type(self):
-        return self._user_account_type
-
-    @property
-    def num_active_projects(self):
-        return self._num_active_projects
-
-    @property
-    def num_remaining_projects(self):
-        return self._num_remaining_projects
+    def to_dict(self) -> Dict[str, Any]:
+        return asdict(self)