Miroslavsimek/be 545 move autoapprove to oarepo workflows

.copyright.tmpl

@@ -0,0 +1,5 @@
+Copyright (C) 2024 CESNET z.s.p.o.
+
+oarepo-workflows is free software; you can redistribute it and/or
+modify it under the terms of the MIT License; see LICENSE file for more
+details.

.coveragerc

@@ -0,0 +1,12 @@
+[report]
+exclude_lines =
+    pragma: no cover
+    ^#
+exclude_also =
+    if TYPE_CHECKING:
+    if __name__ == .__main__.:
+    if TYPE_CHECKING:
+    class .*\bProtocol\):
+    @(abc\.)?abstractmethod
+    raise AssertionError
+    raise NotImplementedError

.github/workflows/build.yaml

@@ -20,7 +20,7 @@ jobs:
       - name: Set up Python
         uses: actions/setup-python@v5
         with:
-          python-version: "3.10"
+          python-version: "3.12"
       - name: Cache pip
         uses: actions/cache@v4
         with:

mypy.ini

@@ -0,0 +1,2 @@
+[mypy]
+disable_error_code = import-untyped, import-not-found

oarepo_workflows/__init__.py

@@ -1,14 +1,25 @@
+#
+# Copyright (C) 2024 CESNET z.s.p.o.
+#
+# oarepo-workflows is free software; you can redistribute it and/or
+# modify it under the terms of the MIT License; see LICENSE file for more
+# details.
+#
+"""Support any workflows on Invenio record."""
+
+from __future__ import annotations
+
 from oarepo_workflows.services.permissions import (
-    AutoApprove,
-    AutoRequest,
-    DefaultWorkflowPermissions,
+    FromRecordWorkflow,
     IfInState,
     WorkflowPermission,
-    WorkflowPermissionPolicy,
+    WorkflowRecordPermissionPolicy,
 )
 
 from .base import Workflow
 from .requests import (
+    AutoApprove,
+    AutoRequest,
     WorkflowRequest,
     WorkflowRequestEscalation,
     WorkflowRequestPolicy,
@@ -19,12 +30,12 @@
     "IfInState",
     "Workflow",
     "WorkflowPermission",
-    "DefaultWorkflowPermissions",
-    "WorkflowPermissionPolicy",
+    "WorkflowRecordPermissionPolicy",
     "WorkflowRequestPolicy",
     "WorkflowRequest",
+    "WorkflowRequestEscalation",
     "WorkflowTransitions",
+    "FromRecordWorkflow",
     "AutoApprove",
     "AutoRequest",
-    "WorkflowRequestEscalation",
 )

oarepo_workflows/base.py

@@ -1,29 +1,119 @@
-import dataclasses
-from typing import Type
+#
+# Copyright (C) 2024 CESNET z.s.p.o.
+#
+# oarepo-workflows is free software; you can redistribute it and/or
+# modify it under the terms of the MIT License; see LICENSE file for more
+# details.
+#
+"""Support any workflows on Invenio record."""
+
+from __future__ import annotations
 
-from flask_babel import LazyString
+import dataclasses
+from typing import TYPE_CHECKING, Any, Protocol
 
-from . import WorkflowPermissionPolicy
+from . import WorkflowRecordPermissionPolicy
 from .requests import WorkflowRequestPolicy
 from .services.permissions import DefaultWorkflowPermissions
 
+if TYPE_CHECKING:
+    from flask_babel import LazyString
+    from flask_principal import Identity
+    from invenio_records_resources.records import Record
+    from invenio_records_resources.services.uow import UnitOfWork
+
 
 @dataclasses.dataclass
 class Workflow:
+    """A workflow definition."""
+
     label: str | LazyString
-    permission_policy_cls: Type[DefaultWorkflowPermissions]
-    request_policy_cls: Type[WorkflowRequestPolicy] = WorkflowRequestPolicy
+    """A human-readable label for the workflow."""
 
-    def permissions(self, action, **over):
-        """Return permission policy for this workflow applicable to the given action."""
+    permission_policy_cls: type[DefaultWorkflowPermissions]
+    """A permission policy class that handles permissions on records that are governed by this workflow."""
+
+    request_policy_cls: type[WorkflowRequestPolicy] = WorkflowRequestPolicy
+    """A request policy class that defines which requests can be applied to records governed by this workflow."""
+
+    def permissions(self, action: str, **over: Any) -> DefaultWorkflowPermissions:
+        """Return permission policy for this workflow applicable to the given action.
+
+        :param  action:      action for which permission is asked
+        """
         return self.permission_policy_cls(action, **over)
 
-    def requests(self):
+    def requests(self) -> WorkflowRequestPolicy:
         """Return instance of request policy for this workflow."""
         return self.request_policy_cls()
 
-    def __post_init__(self):
-        assert not issubclass(self.permission_policy_cls, WorkflowPermissionPolicy)
-        assert issubclass(self.permission_policy_cls, DefaultWorkflowPermissions)
+    def __post_init__(self) -> None:
+        """Check that the classes are subclasses of the expected classes.
 
+        This is just a sanity check to raise an error as soon as possible.
+        """
+        assert not issubclass(
+            self.permission_policy_cls, WorkflowRecordPermissionPolicy
+        )
+        assert issubclass(self.permission_policy_cls, DefaultWorkflowPermissions)
         assert issubclass(self.request_policy_cls, WorkflowRequestPolicy)
+
+
+class StateChangedNotifier(Protocol):
+    """A protocol for a state change notifier.
+
+    State changed notifier is a callable that is called when a state of a record changes,
+    for example as a result of a workflow transition.
+    """
+
+    def __call__(
+        self,
+        identity: Identity,
+        record: Record,
+        previous_state: str,
+        new_state: str,
+        *args: Any,
+        uow: UnitOfWork,
+        **kwargs: Any,
+    ):
+        """Notify about a state change.
+
+        :param identity:        identity of the user who initiated the state change
+        :param record:          record whose state changed
+        :param previous_state:  previous state of the record
+        :param new_state:       new state of the record
+        :param args:            additional arguments
+        :param uow:             unit of work
+        :param kwargs:          additional keyword arguments
+        """
+        ...
+
+
+class WorkflowChangeNotifier(Protocol):
+    """A protocol for a workflow change notifier.
+
+    Workflow changed notifier is a callable that is called
+    when a workflow of a record changes.
+    """
+
+    def __call__(
+        self,
+        identity: Identity,
+        record: Record,
+        previous_workflow_id: str,
+        new_workflow_id: str,
+        *args: Any,
+        uow: UnitOfWork,
+        **kwargs: Any,
+    ):
+        """Notify about a workflow change.
+
+        :param identity:                identity of the user who initiated the workflow change
+        :param record:                  record whose workflow changed
+        :param previous_workflow_id:    previous workflow of the record
+        :param new_workflow_id:         new workflow of the record
+        :param args:                    additional arguments
+        :param uow:                     unit of work
+        :param kwargs:                  additional keyword arguments
+        """
+        ...

oarepo_workflows/errors.py

@@ -1,9 +1,107 @@
+#
+# Copyright (C) 2024 CESNET z.s.p.o.
+#
+# oarepo-workflows is free software; you can redistribute it and/or
+# modify it under the terms of the MIT License; see LICENSE file for more
+# details.
+#
+"""Errors raised by oarepo-workflows."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Optional
+
 from marshmallow import ValidationError
 
+if TYPE_CHECKING:
+    from invenio_records_resources.records import Record
+
+
+def _get_id_from_record(record: Record | dict) -> str:
+    """Get the id from a record.
+
+    :param record: A record or a dict representing a record.
+    :return str: The id of the record.
+    """
+    # community record doesn't have id in dict form, only uuid
+    try:
+        if "id" in record:
+            return str(record["id"])
+    except TypeError:
+        pass
+    if hasattr(record, "id"):
+        return str(record.id)
+    return str(record)
+
+
+def _format_record(record: Record | dict) -> str:
+    """Format a record for error messages.
+
+    :param record: A record or a dict representing a record.
+    :return str: A formatted string representing the record.
+    """
+    return f"{type(record).__name__}[{_get_id_from_record(record)}]"
+
 
 class MissingWorkflowError(ValidationError):
-    """"""
+    """Exception raised when a required workflow is missing."""
+
+    def __init__(self, message: str, record: Record | dict | None = None) -> None:
+        """Initialize the exception."""
+        self.record = record
+        if record:
+            super().__init__(f"{message} Used on record {_format_record(record)}")
+        else:
+            super().__init__(message)
 
 
 class InvalidWorkflowError(ValidationError):
-    """"""
+    """Exception raised when a workflow is invalid."""
+
+    def __init__(
+        self,
+        message: str,
+        record: Record | dict | None = None,
+        community_id: Optional[str] = None,
+    ) -> None:
+        """Initialize the exception."""
+        self.record = record
+        if record:
+            super().__init__(f"{message} Used on record {_format_record(record)}")
+        elif community_id:
+            super().__init__(f"{message} Used on community {community_id}")
+        else:
+            super().__init__(message)
+
+
+class InvalidConfigurationError(Exception):
+    """Exception raised when a configuration is invalid."""
+
+
+class EventTypeNotInWorkflow(Exception):
+    """Exception raised when user tries to create a request with a request type that is not defined in the workflow."""
+
+    def __init__(self, request_type: str, event_type: str, workflow_code: str) -> None:
+        """Initialize the exception."""
+        self.request_type = request_type
+        self.workflow = workflow_code
+        self.event_type = event_type
+
+    @property
+    def description(self) -> str:
+        """Exception's description."""
+        return f"Event type {self.event_type} is not on request type {self.request_type} in workflow {self.workflow}."
+
+
+class RequestTypeNotInWorkflow(Exception):
+    """Exception raised when user tries to create a request with a request type that is not defined in the workflow."""
+
+    def __init__(self, request_type: str, workflow_code: str) -> None:
+        """Initialize the exception."""
+        self.request_type = request_type
+        self.workflow = workflow_code
+
+    @property
+    def description(self) -> str:
+        """Exception's description."""
+        return f"Request type {self.request_type} not in workflow {self.workflow}."