Feature: support for state_timestamp field

oarepo_workflows/records/systemfields/__init__.py

@@ -7,10 +7,11 @@
 #
 """Record layer, system fields."""
 
-from .state import RecordStateField
+from .state import RecordStateField, RecordStateTimestampField
 from .workflow import WorkflowField
 
 __all__ = (
     "RecordStateField",
     "WorkflowField",
+    "RecordStateTimestampField",
 )

oarepo_workflows/records/systemfields/state.py

@@ -9,7 +9,8 @@
 
 from __future__ import annotations
 
-from typing import Any, Optional, Protocol, Self, overload
+from datetime import UTC, datetime
+from typing import Any, Optional, Protocol, Self, cast, overload
 
 from invenio_records.systemfields.base import SystemField
 from oarepo_runtime.records.systemfields import MappingSystemFieldMixin
@@ -26,6 +27,9 @@ class WithState(Protocol):
     state: str
     """State of the record."""
 
+    state_timestamp: datetime
+    """Timestamp of the last state change."""
+
 
 class RecordStateField(MappingSystemFieldMixin, SystemField):
     """State system field."""
@@ -62,9 +66,60 @@ def __get__(
 
     def __set__(self, record: WithState, value: str) -> None:
         """Directly set the state of the record."""
-        self.set_dictkey(record, value)
+        if self.get_dictkey(record) != value:
+            self.set_dictkey(record, value)
+            cast(dict, record)["state_timestamp"] = datetime.now(tz=UTC).isoformat()
+
+    @property
+    def mapping(self) -> dict[str, dict[str, str]]:
+        """Return the opensearch mapping for the state field."""
+        return {
+            self.attr_name: {"type": "keyword"},
+        }
+
+
+class RecordStateTimestampField(MappingSystemFieldMixin, SystemField):
+    """State system field."""
+
+    def __init__(self, key: str = "state_timestamp") -> None:
+        """Initialize the state field."""
+        super().__init__(key=key)
+
+    def post_create(self, record: WithState) -> None:
+        """Set the initial state when record is created."""
+        self.set_dictkey(record, datetime.now(tz=UTC).isoformat())
+
+    def post_init(
+        self, record: WithState, data: dict, model: Optional[Any] = None, **kwargs: Any
+    ) -> None:
+        """Set the initial state when record is created."""
+        if not record.state_timestamp:
+            self.set_dictkey(record, datetime.now(tz=UTC).isoformat())
+
+    @overload
+    def __get__(self, record: None, owner: type | None = None) -> Self: ...
+
+    @overload
+    def __get__(self, record: WithState, owner: type | None = None) -> str: ...
+
+    def __get__(
+        self, record: WithState | None, owner: type | None = None
+    ) -> str | Self:
+        """Get the persistent identifier."""
+        if record is None:
+            return self
+        return self.get_dictkey(record)
 
     @property
     def mapping(self) -> dict[str, dict[str, str]]:
         """Return the opensearch mapping for the state field."""
-        return {self.attr_name: {"type": "keyword"}}
+        # not needed as oarepo-model-builder-workflows already generated this field into the mapping
+        return {
+            self.attr_name: {
+                "type": "date",
+                "format": "strict_date_time||strict_date_time_no_millis||"
+                "basic_date_time||basic_date_time_no_millis||"
+                "basic_date||strict_date||"
+                "strict_date_hour_minute_second||strict_date_hour_minute_second_fraction",
+            },
+        }

run-tests.sh

@@ -29,6 +29,9 @@ pip install -U setuptools pip wheel
 install_python_package oarepo-model-builder
 install_python_package oarepo-model-builder-drafts
 
+# local development
+pip install -e ../oarepo-model-builder-workflows
+
 if test -d thesis ; then
   rm -rf thesis
 fi

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = oarepo-workflows
-version = 1.1.0
+version = 1.1.1
 description =
 authors = Ronald Krist <[email protected]>
 readme = README.md

tests/test_workflow.py

@@ -67,9 +67,15 @@ def test_workflow_publish(users, logged_client, default_workflow_json, search_cl
         ThesisResourceConfig.url_prefix, json=default_workflow_json
     )
     draft_json = create_response.json
-    user_client1.post(
+
+    assert draft_json["state_timestamp"] is not None
+
+    published_json = user_client1.post(
         f"{ThesisResourceConfig.url_prefix}{draft_json['id']}/draft/actions/publish"
-    )
+    ).json
+
+    assert draft_json["state_timestamp"] != published_json["state_timestamp"]
+    assert published_json["state"] == "published"
 
     # in published state, all authenticated users should be able to read, this tests that the preset covers
     # read in all states
@@ -83,6 +89,11 @@ def test_workflow_publish(users, logged_client, default_workflow_json, search_cl
     assert owner_response.status_code == 200
     assert other_response.status_code == 200
 
+    assert owner_response.json["state_timestamp"] == published_json["state_timestamp"]
+    assert owner_response.json["state"] == published_json["state"]
+    assert other_response.json["state_timestamp"] == published_json["state_timestamp"]
+    assert other_response.json["state"] == published_json["state"]
+
 
 def test_query_filter(users, logged_client, default_workflow_json, search_clear):
     user_client1 = logged_client(users[0])