Implement release check events and abort action (#13)

* fix base action

* typo

* new release events

* common dialog data

* adapt schema

* fix dependency

* remove general release events

* add release check events for documents, parts and ecs

* update schema

* remove unneeded constructors

* cdb_obsolete can be None

* Functions can return Actions or list of Actions directly

* make doc

* add documentation for new actions and events

* make schema

* unused import

* remove mentioning workloadresponse

* clarify check events

* change order

* remove some unused code

* make object linking independent of field names

* rename fields to attached_<object>

* adjust documentation to include the attached_<object> fields

---------

Co-authored-by: Jens Kürten <[email protected]>

Author: jens-kuerten

csfunctions/actions/__init__.py

@@ -3,6 +3,8 @@
 from pydantic import Field
 
 from .abort_and_show_error import AbortAndShowErrorAction
+from .base import ActionNames
 from .dummy import DummyAction
 
-Action = Annotated[Union[AbortAndShowErrorAction, DummyAction], Field(discriminator="name")]
+ActionUnion = Union[AbortAndShowErrorAction, DummyAction]
+Action = Annotated[ActionUnion, Field(discriminator="name")]

csfunctions/actions/abort_and_show_error.py

@@ -1,23 +1,10 @@
 from typing import Literal
 
-from pydantic import BaseModel, Field
+from pydantic import Field
 
 from .base import ActionNames, BaseAction
 
 
-class AbortAndShowErrorData(BaseModel):
-    def __init__(self, message: str, **kwargs):
-        super().__init__(message=message, **kwargs)
-
-    message: str = Field("unknown error", description="error message to be shown to the user")
-
-
 class AbortAndShowErrorAction(BaseAction):
-    def __init__(self, id: str, message: str | None = None, **kwargs):  # pylint: disable=redefined-builtin
-        if message:
-            super().__init__(name=ActionNames.ABORT_AND_SHOW_ERROR, id=id, data=AbortAndShowErrorData(message=message))
-        else:
-            super().__init__(name=ActionNames.ABORT_AND_SHOW_ERROR, id=id, data=kwargs["data"])
-
-    name: Literal[ActionNames.ABORT_AND_SHOW_ERROR]
-    data: AbortAndShowErrorData
+    name: Literal[ActionNames.ABORT_AND_SHOW_ERROR] = ActionNames.ABORT_AND_SHOW_ERROR
+    message: str = Field("unknown error", description="error message to be shown to the user")

csfunctions/actions/base.py

@@ -1,6 +1,6 @@
 from enum import Enum
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel
 
 
 class ActionNames(str, Enum):
@@ -10,5 +10,4 @@ class ActionNames(str, Enum):
 
 class BaseAction(BaseModel):
     name: str
-    id: str = Field(..., description="unique identifier")
-    data: dict | None = None
+    id: str | None = None

csfunctions/events/__init__.py

@@ -2,29 +2,39 @@
 
 from pydantic import Field
 
-from .document_release import DocumentReleaseData, DocumentReleaseDialogData, DocumentReleaseEvent
+from .dialog_data import DocumentReleaseDialogData, PartReleaseDialogData
+from .document_release import DocumentReleaseData, DocumentReleaseEvent
+from .document_release_check import DocumentReleaseCheckData, DocumentReleaseCheckEvent
 from .dummy import DummyEvent, DummyEventData
 from .engineering_change_release import EngineeringChangeRelease, EngineeringChangeReleaseData
+from .engineering_change_release_check import EngineeringChangeReleaseCheck, EngineeringChangeReleaseCheckData
 from .field_value_calculation import FieldValueCalculationData, FieldValueCalculationEvent
-from .part_release import PartReleaseData, PartReleaseDialogData, PartReleaseEvent
+from .part_release import PartReleaseData, PartReleaseEvent
+from .part_release_check import PartReleaseCheckData, PartReleaseCheckEvent
 from .workflow_task_trigger import WorkflowTaskTriggerEvent, WorkflowTaskTriggerEventData
 
 Event = Annotated[
     Union[
         DocumentReleaseEvent,
+        DocumentReleaseCheckEvent,
         PartReleaseEvent,
+        PartReleaseCheckEvent,
         FieldValueCalculationEvent,
         DummyEvent,
         EngineeringChangeRelease,
+        EngineeringChangeReleaseCheck,
         WorkflowTaskTriggerEvent,
     ],
     Field(discriminator="name"),
 ]
 EventData = Union[
     DocumentReleaseData,
+    DocumentReleaseCheckData,
     PartReleaseData,
+    PartReleaseCheckData,
     FieldValueCalculationData,
     DummyEventData,
     EngineeringChangeReleaseData,
+    EngineeringChangeReleaseCheckData,
     WorkflowTaskTriggerEventData,
 ]

csfunctions/events/base.py

@@ -6,13 +6,15 @@
 class EventNames(str, Enum):
     DUMMY = "dummy"
     DOCUMENT_RELEASE = "document_release"
+    DOCUMENT_RELEASE_CHECK = "document_release_check"
     PART_RELEASE = "part_release"
+    PART_RELEASE_CHECK = "part_release_check"
     ENGINEERING_CHANGE_RELEASE = "engineering_change_release"
+    ENGINEERING_CHANGE_RELEASE_CHECK = "engineering_change_release_check"
     FIELD_VALUE_CALCULATION = "field_value_calculation"
     WORKFLOW_TASK_TRIGGER = "workflow_task_trigger"
 
 
 class BaseEvent(BaseModel):
     name: str
     event_id: str = Field(..., description="unique identifier")
-    data: dict | None = None

csfunctions/events/dialog_data.py

@@ -0,0 +1,15 @@
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+
+class DocumentReleaseDialogData(BaseModel):
+    dialog_type: Literal["document_release"] = "document_release"
+    cdbprot_remark: str | None = Field(None, description="remark")
+    cdb_ec_id: str | None = Field(None, description="Engineering Change ID")
+
+
+class PartReleaseDialogData(BaseModel):
+    dialog_type: Literal["part_release"] = "part_release"
+    cdbprot_remark: str | None = Field("", description="remark")
+    cdb_ec_id: str | None = Field("", description="Engineering Change ID")

csfunctions/events/document_release.py

@@ -5,25 +5,15 @@
 from csfunctions.objects import Document, Part
 
 from .base import BaseEvent, EventNames
-
-
-class DocumentReleaseDialogData(BaseModel):
-    cdbprot_remark: str | None = Field(None, description="remark")
-    cdb_ec_id: str | None = Field(None, description="Engineering Change ID")
+from .dialog_data import DocumentReleaseDialogData
 
 
 class DocumentReleaseData(BaseModel):
-    def __init__(self, documents: list[Document], parts: list[Part], dialog_data: dict, **kwargs):
-        super().__init__(documents=documents, parts=parts, dialog_data=dialog_data, **kwargs)
-
-    documents: list[Document] = Field(..., description="List if documents that were released.")
+    documents: list[Document] = Field(..., description="List of documents that were released.")
     parts: list[Part] = Field(..., description="List of parts that belong to the released documents")
     dialog_data: DocumentReleaseDialogData
 
 
 class DocumentReleaseEvent(BaseEvent):
-    def __init__(self, event_id: str, data: DocumentReleaseData, **_):
-        super().__init__(name=EventNames.DOCUMENT_RELEASE, event_id=event_id, data=data)
-
-    name: Literal[EventNames.DOCUMENT_RELEASE]
+    name: Literal[EventNames.DOCUMENT_RELEASE] = EventNames.DOCUMENT_RELEASE
     data: DocumentReleaseData

csfunctions/events/document_release_check.py

@@ -0,0 +1,19 @@
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+from csfunctions.objects import Document, Part
+
+from .base import BaseEvent, EventNames
+from .dialog_data import DocumentReleaseDialogData
+
+
+class DocumentReleaseCheckData(BaseModel):
+    documents: list[Document] = Field(..., description="List of documents that will be released.")
+    attached_parts: list[Part] = Field(..., description="List of parts that belong to the documents")
+    dialog_data: DocumentReleaseDialogData
+
+
+class DocumentReleaseCheckEvent(BaseEvent):
+    name: Literal[EventNames.DOCUMENT_RELEASE_CHECK] = EventNames.DOCUMENT_RELEASE_CHECK
+    data: DocumentReleaseCheckData

csfunctions/events/engineering_change_release.py

@@ -16,8 +16,5 @@ class EngineeringChangeReleaseData(BaseModel):
 
 
 class EngineeringChangeRelease(BaseEvent):
-    def __init__(self, event_id: str, data: EngineeringChangeReleaseData, **_):
-        super().__init__(name=EventNames.ENGINEERING_CHANGE_RELEASE, event_id=event_id, data=data)
-
-    name: Literal[EventNames.ENGINEERING_CHANGE_RELEASE]
+    name: Literal[EventNames.ENGINEERING_CHANGE_RELEASE] = EventNames.ENGINEERING_CHANGE_RELEASE
     data: EngineeringChangeReleaseData

csfunctions/events/engineering_change_release_check.py

@@ -0,0 +1,20 @@
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+from csfunctions.objects import Document, EngineeringChange, Part
+
+from .base import BaseEvent, EventNames
+
+
+class EngineeringChangeReleaseCheckData(BaseModel):
+    attached_documents: list[Document] = Field(..., description="List of included documents.")
+    attached_parts: list[Part] = Field(..., description="List of included parts.")
+    engineering_changes: list[EngineeringChange] = Field(
+        ..., description="List of engineering changes that will be released."
+    )
+
+
+class EngineeringChangeReleaseCheck(BaseEvent):
+    name: Literal[EventNames.ENGINEERING_CHANGE_RELEASE_CHECK] = EventNames.ENGINEERING_CHANGE_RELEASE_CHECK
+    data: EngineeringChangeReleaseCheckData