IWF-357: Refactor

iwf/command_results.py

@@ -2,7 +2,7 @@
 from dataclasses import dataclass
 from typing import Any, Union
 
-from iwf.errors import WorkflowDefinitionError
+from iwf.errors import WorkflowDefinitionError, NotRegisteredError
 from iwf.iwf_api.models import (
     ChannelRequestStatus,
     CommandResults as IdlCommandResults,
@@ -59,12 +59,13 @@ def from_idl_command_results(
 
     if not isinstance(idl_results.inter_state_channel_results, Unset):
         for inter in idl_results.inter_state_channel_results:
-            val_type = internal_channel_types.get_type(inter.channel_name)
 
-            if val_type is None:
+            try:
+                val_type = internal_channel_types.get_type(inter.channel_name)
+            except NotRegisteredError as exception:
                 raise WorkflowDefinitionError(
                     "internal channel is not registered: " + inter.channel_name
-                )
+                ) from exception
 
             encoded = object_encoder.decode(inter.value, val_type)
 

iwf/communication.py

@@ -1,6 +1,6 @@
 from typing import Any, Optional, Union
 
-from iwf.errors import WorkflowDefinitionError
+from iwf.errors import WorkflowDefinitionError, NotRegisteredError
 from iwf.iwf_api.models import (
     EncodedObject,
     InterStateChannelPublishing,
@@ -48,12 +48,12 @@ def trigger_state_execution(self, state: Union[str, type], state_input: Any = No
         self._state_movements.append(movement)
 
     def publish_to_internal_channel(self, channel_name: str, value: Any = None):
-        registered_type = self._internal_channel_type_store.get_type(channel_name)
-
-        if registered_type is None:
+        try:
+            registered_type = self._internal_channel_type_store.get_type(channel_name)
+        except NotRegisteredError as exception:
             raise WorkflowDefinitionError(
                 f"InternalChannel channel_name is not defined {channel_name}"
-            )
+            ) from exception
 
         if (
             value is not None

iwf/errors.py

@@ -18,6 +18,10 @@ class InvalidArgumentError(Exception):
     pass
 
 
+class NotRegisteredError(Exception):
+    pass
+
+
 class HttpError(RuntimeError):
     def __init__(self, status: int, err_resp: ErrorResponse):
         super().__init__(err_resp.detail)

iwf/type_store.py

@@ -2,7 +2,7 @@
 from enum import Enum
 
 from iwf.communication_schema import CommunicationMethod
-from iwf.errors import WorkflowDefinitionError
+from iwf.errors import WorkflowDefinitionError, NotRegisteredError
 
 
 class Type(Enum):
@@ -26,11 +26,11 @@ def is_valid_name_or_prefix(self, name: str) -> bool:
         t = self._do_get_type(name)
         return t is not None
 
-    def get_type(self, name: str) -> Optional[type]:
+    def get_type(self, name: str) -> type:
         t = self._do_get_type(name)
 
         if t is None:
-            raise ValueError(f"{self._class_type} not registered: {name}")
+            raise NotRegisteredError(f"{self._class_type} not registered: {name}")
 
         return t