Merge branch 'main' of github.com:cslab/functions-sdk-python into custom-operations

Author: Jens Kürten

README.md

@@ -12,7 +12,7 @@ Functions are deeply integrated in the CIM Database Cloud Webhooks technology. T
 
 Python 3.10+
 
-csfunctions is build with [Pydantic 2](https://docs.pydantic.dev/latest/)
+csfunctions is built with [Pydantic 2](https://docs.pydantic.dev/latest/)
 
 ## Installation
 Install using pip:

csfunctions/actions/__init__.py

@@ -5,8 +5,9 @@
 from .abort_and_show_error import AbortAndShowErrorAction
 from .base import ActionNames
 from .dummy import DummyAction
+from .start_workflow import StartWorkflowAction
 
-ActionUnion = Union[AbortAndShowErrorAction, DummyAction]
+ActionUnion = Union[AbortAndShowErrorAction, DummyAction, StartWorkflowAction]
 Action = Annotated[ActionUnion, Field(discriminator="name")]
 
 __all__ = [
@@ -15,4 +16,5 @@
     "DummyAction",
     "AbortAndShowErrorAction",
     "ActionUnion",
+    "StartWorkflowAction",
 ]

csfunctions/actions/base.py

@@ -6,6 +6,7 @@
 class ActionNames(str, Enum):
     ABORT_AND_SHOW_ERROR = "abort_and_show_error"
     DUMMY = "dummy"
+    START_WORKFLOW = "start_workflow"
 
 
 class BaseAction(BaseModel):

csfunctions/actions/start_workflow.py

@@ -0,0 +1,53 @@
+from typing import Literal, Optional
+
+from pydantic import BaseModel, Field
+
+from .base import ActionNames, BaseAction
+
+
+class Subject(BaseModel):
+    subject_id: str = Field(..., description="ID of the subject, eg. a role name or personalnummer")
+    subject_type: Literal["Person", "PCS Role", "Common Role"] = Field(
+        ..., description="Type of the subject: Person, PCS Role or Common Role"
+    )
+
+
+class TaskConfiguration(BaseModel):
+    task_id: str = Field(..., description="Identifier for the task")
+    responsible: Optional[Subject] = Field(default=None, description="Responsible subject for the task")
+    recipients: list[Subject] = Field(
+        default_factory=list,
+        description="List of recipients for the task (only used by information tasks)",
+    )
+    description: str | None = Field(
+        default=None,
+        description="Description of the task. If not set, the existing description will be kept. "
+        "(max. 1024 characters)",
+        max_length=1024,
+    )
+    title: str | None = Field(
+        default=None,
+        description="Title of the task. If not set, the existing title will be kept. (max. 60 characters)",
+        max_length=60,
+    )
+
+
+class StartWorkflowAction(BaseAction):
+    name: Literal[ActionNames.START_WORKFLOW] = ActionNames.START_WORKFLOW
+    template_id: str = Field(..., description="ID of the workflow template to start")
+    cdb_project_id: str | None = Field(
+        default=None,
+        description="ID of the project in which the workflow should be started",
+    )
+    title: str = Field(..., description="Title of the workflow (max. 255 characters)", max_length=255)
+    attachment_ids: list[str] = Field(
+        default_factory=list,
+        description="List of cdb_object_ids to attach to the workflow",
+    )
+    global_briefcase_object_ids: list[str] = Field(
+        default_factory=list,
+        description="List of cdb_object_ids to attach to the global briefcase",
+    )
+    task_configurations: list[TaskConfiguration] = Field(
+        default_factory=list, description="List of task configurations"
+    )

csfunctions/events/__init__.py

@@ -17,6 +17,11 @@
 from .dummy import DummyEvent, DummyEventData
 from .engineering_change_release_check import EngineeringChangeReleaseCheckData, EngineeringChangeReleaseCheckEvent
 from .engineering_change_released import EngineeringChangeReleasedData, EngineeringChangeReleasedEvent
+from .engineering_change_status_change_check import (
+    EngineeringChangeStatusChangeCheckData,
+    EngineeringChangeStatusChangeCheckEvent,
+)
+from .engineering_change_status_changed import EngineeringChangeStatusChangedData, EngineeringChangeStatusChangedEvent
 from .field_value_calculation import FieldValueCalculationData, FieldValueCalculationEvent
 from .part_create_check import PartCreateCheckData, PartCreateCheckEvent
 from .part_field_calculation import PartFieldCalculationData, PartFieldCalculationEvent
@@ -37,6 +42,8 @@
         DummyEvent,
         EngineeringChangeReleasedEvent,
         EngineeringChangeReleaseCheckEvent,
+        EngineeringChangeStatusChangedEvent,
+        EngineeringChangeStatusChangeCheckEvent,
         WorkflowTaskTriggerEvent,
         DocumentCreateCheckEvent,
         DocumentModifyCheckEvent,
@@ -58,6 +65,8 @@
     DummyEventData,
     EngineeringChangeReleasedData,
     EngineeringChangeReleaseCheckData,
+    EngineeringChangeStatusChangedData,
+    EngineeringChangeStatusChangeCheckData,
     WorkflowTaskTriggerEventData,
     DocumentCreateCheckData,
     DocumentModifyCheckData,
@@ -88,6 +97,8 @@
     "DummyEventData",
     "EngineeringChangeReleasedData",
     "EngineeringChangeReleaseCheckData",
+    "EngineeringChangeStatusChangedData",
+    "EngineeringChangeStatusChangeCheckData",
     "WorkflowTaskTriggerEventData",
     "DocumentReleasedDialogData",
     "PartReleasedDialogData",

csfunctions/events/base.py

@@ -19,6 +19,8 @@ class EventNames(str, Enum):
     DOCUMENT_MODIFY_CHECK = "document_modify_check"
     PART_CREATE_CHECK = "part_create_check"
     PART_MODIFY_CHECK = "part_modify_check"
+    ENGINEERING_CHANGE_STATUS_CHANGED = "engineering_change_status_changed"
+    ENGINEERING_CHANGE_STATUS_CHANGE_CHECK = "engineering_change_status_change_check"
     CUSTOM_OPERATION_DOCUMENT = "custom_operation_document"
     CUSTOM_OPERATION_PART = "custom_operation_part"
 

csfunctions/events/engineering_change_status_change_check.py

@@ -0,0 +1,21 @@
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+from csfunctions.objects import Document, EngineeringChange, Part
+
+from .base import BaseEvent, EventNames
+
+
+class EngineeringChangeStatusChangeCheckData(BaseModel):
+    engineering_change: EngineeringChange = Field(
+        ..., description="The engineering change that will have its status modified"
+    )
+    target_status: int = Field(..., description="The target status of the engineering change")
+    documents: list[Document] = Field(..., description="List of documents attached to the engineering change")
+    parts: list[Part] = Field(..., description="List of parts attached to the engineering change")
+
+
+class EngineeringChangeStatusChangeCheckEvent(BaseEvent):
+    name: Literal[EventNames.ENGINEERING_CHANGE_STATUS_CHANGE_CHECK] = EventNames.ENGINEERING_CHANGE_STATUS_CHANGE_CHECK
+    data: EngineeringChangeStatusChangeCheckData

csfunctions/events/engineering_change_status_changed.py

@@ -0,0 +1,21 @@
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+from csfunctions.objects import Document, EngineeringChange, Part
+
+from .base import BaseEvent, EventNames
+
+
+class EngineeringChangeStatusChangedData(BaseModel):
+    engineering_change: EngineeringChange = Field(
+        ..., description="The engineering change that had its status modified"
+    )
+    prev_status: int = Field(..., description="The previous status of the engineering change")
+    documents: list[Document] = Field(..., description="List of documents attached to the engineering change")
+    parts: list[Part] = Field(..., description="List of parts attached to the engineering change")
+
+
+class EngineeringChangeStatusChangedEvent(BaseEvent):
+    name: Literal[EventNames.ENGINEERING_CHANGE_STATUS_CHANGED] = EventNames.ENGINEERING_CHANGE_STATUS_CHANGED
+    data: EngineeringChangeStatusChangedData

csfunctions/handler.py

@@ -1,4 +1,5 @@
 import json
+import logging
 import os
 import sys
 import traceback
@@ -16,6 +17,8 @@
 from csfunctions.response import ResponseUnion
 from csfunctions.service import Service
 
+logger = logging.getLogger(__name__)
+
 
 class FunctionNotRegistered(ValueError):
     """
@@ -68,15 +71,17 @@ def link_objects(event: Event):
     # e.g. all parts would be in Event.data.parts = list[Part]
 
     for field_name in data.model_fields_set:
-        # go through each field in data and look for fields that are lists
-        # objects will always be passed in a list
+        # go through each field in data and look for fields that are lists of BaseObjects
+        # or direct BaseObjects
 
         field = getattr(data, field_name)
         if isinstance(field, list):
             for obj in field:
                 # the list might contain entries that are not objects, so we check first
                 if isinstance(obj, BaseObject):
                     obj.link_objects(data)
+        elif isinstance(field, BaseObject):
+            field.link_objects(data)
 
 
 def execute(function_name: str, request_body: str, function_dir: str = "src") -> str:
@@ -115,6 +120,7 @@ def execute(function_name: str, request_body: str, function_dir: str = "src") ->
         response.event_id = request.event.event_id
 
     except Exception as e:  # pylint: disable=broad-except
+        logger.error("An error occurred while executing function %s", function_name, exc_info=True)
         response = ErrorResponse(message=str(e), error_type=type(e).__name__, trace=traceback.format_exc(), id="")
 
     return response.model_dump_json()