Package refator and handle process item signal

kuflow-temporal-activity-kuflow/kuflow_temporal_activity_kuflow/__init__.py

@@ -22,10 +22,8 @@
 # SOFTWARE.
 #
 
-from .kuflow_activities import KuFlowActivities
+from ._activities import KuFlowActivities
 
 
-KUFLOW_ENGINE_SIGNAL_COMPLETED_TASK = "KuFlow_Engine_Signal_Completed_Task"
-
-__all__ = ["KuFlowActivities", "KUFLOW_ENGINE_SIGNAL_COMPLETED_TASK"]
+__all__ = ["KuFlowActivities"]
 __version__ = "2.0.0.dev0"

kuflow-temporal-activity-kuflow/kuflow_temporal_activity_kuflow/kuflow_activities.py → kuflow-temporal-activity-kuflow/kuflow_temporal_activity_kuflow/_activities.py

@@ -26,11 +26,10 @@
 
 from kuflow_rest import KuFlowRestClient
 from kuflow_rest import models as models_rest
-from kuflow_temporal_common import converter, exceptions
+from kuflow_temporal_common import create_application_error
 
 from . import _validation as validation
 from . import models as models_temporal
-from .converter import KuFlowComposableEncodingPayloadConverter
 
 
 class KuFlowActivities:
@@ -58,7 +57,6 @@ def __init__(self, kuflow_client: KuFlowRestClient) -> None:
         ]
 
     @activity.defn(name="KuFlow_Engine_retrievePrincipal")
-    @converter.register(encoding_payload_converter_class=KuFlowComposableEncodingPayloadConverter)
     async def retrieve_principal(
         self,
         request: models_temporal.PrincipalRetrieveRequest,
@@ -70,10 +68,9 @@ async def retrieve_principal(
 
             return models_temporal.PrincipalRetrieveResponse(principal=principal)
         except Exception as err:
-            raise exceptions.create_application_error(err) from err
+            raise create_application_error(err) from err
 
     @activity.defn(name="KuFlow_Engine_retrieveTenantUser")
-    @converter.register(encoding_payload_converter_class=KuFlowComposableEncodingPayloadConverter)
     async def retrieve_tenant_user(
         self,
         request: models_temporal.TenantUserRetrieveRequest,
@@ -85,10 +82,9 @@ async def retrieve_tenant_user(
 
             return models_temporal.TenantUserRetrieveResponse(tenant_user=tenant_user)
         except Exception as err:
-            raise exceptions.create_application_error(err) from err
+            raise create_application_error(err) from err
 
     @activity.defn(name="KuFlow_Engine_findProcesses")
-    @converter.register(encoding_payload_converter_class=KuFlowComposableEncodingPayloadConverter)
     async def find_processes(
         self,
         request: models_temporal.ProcesFindRequest,
@@ -100,10 +96,9 @@ async def find_processes(
 
             return models_temporal.ProcesFindResponse(processes=proces_page)
         except Exception as err:
-            raise exceptions.create_application_error(err) from err
+            raise create_application_error(err) from err
 
     @activity.defn(name="KuFlow_Engine_retrieveProcess")
-    @converter.register(encoding_payload_converter_class=KuFlowComposableEncodingPayloadConverter)
     async def retrieve_process(
         self,
         request: models_temporal.ProcessRetrieveRequest,
@@ -115,10 +110,9 @@ async def retrieve_process(
 
             return models_temporal.ProcessRetrieveResponse(process=process)
         except Exception as err:
-            raise exceptions.create_application_error(err) from err
+            raise create_application_error(err) from err
 
     @activity.defn(name="KuFlow_Engine_updateProcessEntity")
-    @converter.register(encoding_payload_converter_class=KuFlowComposableEncodingPayloadConverter)
     async def update_process_entity(
         self,
         request: models_temporal.ProcessEntityUpdateRequest,
@@ -134,10 +128,9 @@ async def update_process_entity(
 
             return models_temporal.ProcessEntityUpdateResponse(process=process)
         except Exception as err:
-            raise exceptions.create_application_error(err) from err
+            raise create_application_error(err) from err
 
     @activity.defn(name="KuFlow_Engine_patchProcessEntity")
-    @converter.register(encoding_payload_converter_class=KuFlowComposableEncodingPayloadConverter)
     async def patch_process_entity(
         self,
         request: models_temporal.ProcessEntityPatchRequest,
@@ -151,10 +144,9 @@ async def patch_process_entity(
 
             return models_temporal.ProcessEntityPatchResponse(process=process)
         except Exception as err:
-            raise exceptions.create_application_error(err) from err
+            raise create_application_error(err) from err
 
     @activity.defn(name="KuFlow_Engine_updateProcessMetadata")
-    @converter.register(encoding_payload_converter_class=KuFlowComposableEncodingPayloadConverter)
     async def update_process_metadata(
         self,
         request: models_temporal.ProcessMetadataUpdateRequest,
@@ -170,10 +162,9 @@ async def update_process_metadata(
 
             return models_temporal.ProcessMetadataUpdateResponse(process=process)
         except Exception as err:
-            raise exceptions.create_application_error(err) from err
+            raise create_application_error(err) from err
 
     @activity.defn(name="KuFlow_Engine_patchProcessMetadata")
-    @converter.register(encoding_payload_converter_class=KuFlowComposableEncodingPayloadConverter)
     async def patch_process_metadata(
         self,
         request: models_temporal.ProcessMetadataPatchRequest,
@@ -187,10 +178,9 @@ async def patch_process_metadata(
 
             return models_temporal.ProcessMetadataPatchResponse(process=process)
         except Exception as err:
-            raise exceptions.create_application_error(err) from err
+            raise create_application_error(err) from err
 
     @activity.defn(name="KuFlow_Engine_changeProcessInitiator")
-    @converter.register(encoding_payload_converter_class=KuFlowComposableEncodingPayloadConverter)
     async def change_process_initiator(
         self,
         request: models_temporal.ProcessInitiatorChangeRequest,
@@ -208,10 +198,9 @@ async def change_process_initiator(
 
             return models_temporal.ProcessInitiatorChangeResponse(process=process)
         except Exception as err:
-            raise exceptions.create_application_error(err) from err
+            raise create_application_error(err) from err
 
     @activity.defn(name="KuFlow_Engine_findProcessItems")
-    @converter.register(encoding_payload_converter_class=KuFlowComposableEncodingPayloadConverter)
     async def find_process_items(
         self,
         request: models_temporal.ProcessItemFindRequest,
@@ -224,10 +213,9 @@ async def find_process_items(
 
             return models_temporal.ProcessItemFindResponse(process_items=process_items)
         except Exception as err:
-            raise exceptions.create_application_error(err) from err
+            raise create_application_error(err) from err
 
     @activity.defn(name="KuFlow_Engine_retrieveProcessItem")
-    @converter.register(encoding_payload_converter_class=KuFlowComposableEncodingPayloadConverter)
     async def retrieve_process_item(
         self,
         request: models_temporal.ProcessItemRetrieveRequest,
@@ -239,10 +227,9 @@ async def retrieve_process_item(
 
             return models_temporal.ProcessItemRetrieveResponse(process_item=process_item)
         except Exception as err:
-            raise exceptions.create_application_error(err) from err
+            raise create_application_error(err) from err
 
     @activity.defn(name="KuFlow_Engine_createProcessItem")
-    @converter.register(encoding_payload_converter_class=KuFlowComposableEncodingPayloadConverter)
     async def create_process_item(
         self,
         request: models_temporal.ProcessItemCreateRequest,
@@ -263,10 +250,9 @@ async def create_process_item(
 
             return models_temporal.ProcessItemCreateResponse(process_item=process_item)
         except Exception as err:
-            raise exceptions.create_application_error(err) from err
+            raise create_application_error(err) from err
 
     @activity.defn(name="KuFlow_Engine_completeProcessItemTask")
-    @converter.register(encoding_payload_converter_class=KuFlowComposableEncodingPayloadConverter)
     async def complete_process_item_task(
         self,
         request: models_temporal.ProcessItemTaskCompleteRequest,
@@ -278,10 +264,9 @@ async def complete_process_item_task(
 
             return models_temporal.ProcessItemTaskCompleteResponse(process_item=process_item)
         except Exception as err:
-            raise exceptions.create_application_error(err) from err
+            raise create_application_error(err) from err
 
     @activity.defn(name="KuFlow_Engine_claimProcessItemTask")
-    @converter.register(encoding_payload_converter_class=KuFlowComposableEncodingPayloadConverter)
     async def claim_process_item_task(
         self,
         request: models_temporal.ProcessItemTaskClaimRequest,
@@ -293,10 +278,9 @@ async def claim_process_item_task(
 
             return models_temporal.ProcessItemTaskClaimResponse(process_item=process_item)
         except Exception as err:
-            raise exceptions.create_application_error(err) from err
+            raise create_application_error(err) from err
 
     @activity.defn(name="KuFlow_Engine_assignProcessItemTask")
-    @converter.register(encoding_payload_converter_class=KuFlowComposableEncodingPayloadConverter)
     async def assign_process_item_task(
         self,
         request: models_temporal.ProcessItemTaskAssignRequest,
@@ -312,10 +296,9 @@ async def assign_process_item_task(
 
             return models_temporal.ProcessItemTaskAssignResponse(process_item=process_item)
         except Exception as err:
-            raise exceptions.create_application_error(err) from err
+            raise create_application_error(err) from err
 
     @activity.defn(name="KuFlow_Engine_updateProcessItemTaskData")
-    @converter.register(encoding_payload_converter_class=KuFlowComposableEncodingPayloadConverter)
     async def update_process_item_task_data(
         self,
         request: models_temporal.ProcessItemTaskDataUpdateRequest,
@@ -331,10 +314,9 @@ async def update_process_item_task_data(
 
             return models_temporal.ProcessItemTaskDataUpdateResponse(process_item=process_item)
         except Exception as err:
-            raise exceptions.create_application_error(err) from err
+            raise create_application_error(err) from err
 
     @activity.defn(name="KuFlow_Engine_patchProcessItemTaskData")
-    @converter.register(encoding_payload_converter_class=KuFlowComposableEncodingPayloadConverter)
     async def patch_process_item_task_data(
         self,
         request: models_temporal.ProcessItemTaskDataPatchRequest,
@@ -348,10 +330,9 @@ async def patch_process_item_task_data(
 
             return models_temporal.ProcessItemTaskDataPatchResponse(process_item=process_item)
         except Exception as err:
-            raise exceptions.create_application_error(err) from err
+            raise create_application_error(err) from err
 
     @activity.defn(name="KuFlow_Engine_appendProcessItemTaskLog")
-    @converter.register(encoding_payload_converter_class=KuFlowComposableEncodingPayloadConverter)
     async def append_process_item_task_log(
         self,
         request: models_temporal.ProcessItemTaskLogAppendRequest,
@@ -367,4 +348,4 @@ async def append_process_item_task_log(
 
             return models_temporal.ProcessItemTaskLogAppendResponse(process_item=process_item)
         except Exception as err:
-            raise exceptions.create_application_error(err) from err
+            raise create_application_error(err) from err

kuflow-temporal-activity-kuflow/kuflow_temporal_activity_kuflow/_validation.py

@@ -25,7 +25,7 @@
 from temporalio.exceptions import ApplicationError
 
 from kuflow_rest import models as models_rest
-from kuflow_temporal_common.exceptions import KuFlowFailureType
+from kuflow_temporal_common import KuFlowFailureType
 
 from . import models as models_temporal
 

kuflow-temporal-activity-kuflow/kuflow_temporal_activity_kuflow/converter.py

@@ -25,24 +25,18 @@
 import json
 from typing import Any, Optional, Type
 
-from temporalio import workflow
 from temporalio.api.common.v1 import Payload
 from temporalio.converter import (
-    CompositePayloadConverter,
-    DefaultPayloadConverter,
     EncodingPayloadConverter,
     JSONPlainPayloadConverter,
 )
 
+from kuflow_rest import Deserializer, Model, Serializer
 from kuflow_rest import models as models_rest
 
 from . import models as models_temporal
 
 
-with workflow.unsafe.imports_passed_through():
-    from kuflow_rest import Deserializer, Model, Serializer
-
-
 class KuFlowComposableEncodingPayloadConverter(EncodingPayloadConverter):
     def __init__(self, default_json_converter=JSONPlainPayloadConverter()) -> None:
         self._default_json_converter = default_json_converter
@@ -69,47 +63,3 @@ def from_payload(self, payload: Payload, type_hint: Optional[Type] = None) -> An
 
         as_python_object = json.loads(payload.data)
         return self._deserialize(type_hint.__name__, as_python_object)
-
-
-class KuFlowEncodingPayloadConverter(EncodingPayloadConverter):
-    def __init__(self, default_json_converter=JSONPlainPayloadConverter()) -> None:
-        self._default_json_converter = default_json_converter
-        client_models_rest = {k: v for k, v in models_rest.__dict__.items() if isinstance(v, type)}
-        client_models_temporal = {k: v for k, v in models_temporal.__dict__.items() if isinstance(v, type)}
-        client_models = {**client_models_rest, **client_models_temporal}
-        self._serialize = Serializer(client_models)
-        self._deserialize = Deserializer(client_models)
-
-    @property
-    def encoding(self) -> str:
-        return "json/plain"
-
-    def to_payload(self, value: Any) -> Optional[Payload]:
-        if isinstance(value, Model):
-            serialized = self._serialize.body(value, value.__class__.__name__)
-            return self._default_json_converter.to_payload(serialized)
-        else:
-            return self._default_json_converter.to_payload(value)
-
-    def from_payload(self, payload: Payload, type_hint: Optional[Type] = None) -> Any:
-        # Only supports json/plain. You need to modify this converter in order tu support encryption
-        message_type = payload.metadata.get("encoding", b"<unknown>").decode()
-        if message_type != "json/plain":
-            return self._default_json_converter.from_payload(payload.data, type_hint)
-
-        if issubclass(type_hint, Model):
-            as_python_object = json.loads(payload.data)
-            return self._deserialize(type_hint.__name__, as_python_object)
-        else:
-            return self._default_json_converter.from_payload(payload, type_hint)
-
-
-class KuFlowPayloadConverter(CompositePayloadConverter):
-    def __init__(self) -> None:
-        #  Replace default JSONPlainPayloadConverter with a custom converter for KuFlow
-        super().__init__(
-            *[
-                c if not isinstance(c, JSONPlainPayloadConverter) else KuFlowEncodingPayloadConverter()
-                for c in DefaultPayloadConverter.default_encoding_payload_converters
-            ]
-        )

kuflow-temporal-activity-kuflow/kuflow_temporal_activity_kuflow/models/__init__.py

@@ -57,14 +57,8 @@
     ProcessMetadataUpdateResponse,
     ProcessRetrieveRequest,
     ProcessRetrieveResponse,
-    RobotWorkflowRequest,
-    RobotWorkflowResponse,
     TenantUserRetrieveRequest,
     TenantUserRetrieveResponse,
-    UserActionWorkflowRequest,
-    UserActionWorkflowResponse,
-    WorkflowRequest,
-    WorkflowResponse,
 )
 
 
@@ -103,12 +97,15 @@
     "ProcessMetadataUpdateResponse",
     "ProcessRetrieveRequest",
     "ProcessRetrieveResponse",
-    "RobotWorkflowRequest",
-    "RobotWorkflowResponse",
     "TenantUserRetrieveRequest",
     "TenantUserRetrieveResponse",
-    "UserActionWorkflowRequest",
-    "UserActionWorkflowResponse",
-    "WorkflowRequest",
-    "WorkflowResponse",
 ]
+
+from kuflow_rest import models as models_rest
+from kuflow_temporal_common import register_serializable_models
+
+from .. import models as models_activity
+
+
+register_serializable_models(models_rest.__dict__)
+register_serializable_models(models_activity.__dict__)