Fix mypy errors

microsoft · Nov 14, 2024 · ba87c48 · ba87c48
1 parent fc9154f
commit ba87c48
Show file tree

Hide file tree

Showing 5 changed files with 28 additions and 31 deletions.
diff --git a/python/semantic_kernel/processes/dapr_runtime/dapr_kernel_process.py b/python/semantic_kernel/processes/dapr_runtime/dapr_kernel_process.py
@@ -31,15 +31,16 @@ async def start(
     if initial_event is None:
         raise ProcessInvalidConfigurationException("initial_event cannot be None")
 
-    if isinstance(initial_event, Enum):
-        initial_event = initial_event.value
+    initial_event_str: str | KernelProcessEvent = (
+        initial_event.value if isinstance(initial_event, Enum) else initial_event
+    )
 
-    if isinstance(initial_event, str):
-        initial_event = KernelProcessEvent(id=initial_event, data=kwargs.get("data"))
+    if isinstance(initial_event_str, str):
+        initial_event_str = KernelProcessEvent(id=initial_event_str, data=kwargs.get("data"))
 
     if process_id is not None:
         process.state.id = process_id
 
     process_context = DaprKernelProcessContext(process=process)
-    await process_context.start_with_event(initial_event)
+    await process_context.start_with_event(initial_event_str)
     return process_context
diff --git a/python/semantic_kernel/processes/local_runtime/local_kernel_process.py b/python/semantic_kernel/processes/local_runtime/local_kernel_process.py
@@ -27,12 +27,13 @@ async def start(
     if initial_event is None:
         raise ProcessInvalidConfigurationException("initial_event cannot be None")
 
-    if isinstance(initial_event, Enum):
-        initial_event = initial_event.value
+    initial_event_str: str | KernelProcessEvent = (
+        initial_event.value if isinstance(initial_event, Enum) else initial_event
+    )
 
-    if isinstance(initial_event, str):
-        initial_event = KernelProcessEvent(id=initial_event, data=kwargs.get("data"))
+    if isinstance(initial_event_str, str):
+        initial_event_str = KernelProcessEvent(id=initial_event_str, data=kwargs.get("data"))
 
     process_context = LocalKernelProcessContext(process, kernel)
-    await process_context.start_with_event(initial_event)
+    await process_context.start_with_event(initial_event_str)
     return process_context
diff --git a/python/semantic_kernel/processes/process_builder.py b/python/semantic_kernel/processes/process_builder.py
@@ -76,16 +76,15 @@ def resolve_function_target(
 
     def where_input_event_is(self, event_id: str | Enum) -> "ProcessFunctionTargetBuilder":
         """Filters the input event."""
-        if isinstance(event_id, Enum):
-            event_id = event_id.value
+        event_id_str: str = event_id.value if isinstance(event_id, Enum) else event_id
 
-        if event_id not in self.external_event_target_map:
-            raise ValueError(f"The process named '{self.name}' does not expose an event with Id '{event_id}'")
+        if event_id_str not in self.external_event_target_map:
+            raise ValueError(f"The process named '{self.name}' does not expose an event with Id '{event_id_str}'")
 
-        target = self.external_event_target_map[event_id]
+        target = self.external_event_target_map[event_id_str]
         target = copy(target)
         target.step = self
-        target.target_event_id = event_id
+        target.target_event_id = event_id_str
         return target
 
     def on_input_event(self, event_id: str | Enum) -> "ProcessEdgeBuilder":  # type: ignore
@@ -94,10 +93,9 @@ def on_input_event(self, event_id: str | Enum) -> "ProcessEdgeBuilder":  # type:
 
         ProcessEdgeBuilder.model_rebuild()
 
-        if isinstance(event_id, Enum):
-            event_id = event_id.value
+        event_id_str: str = event_id.value if isinstance(event_id, Enum) else event_id
 
-        return ProcessEdgeBuilder(source=self, event_id=event_id)
+        return ProcessEdgeBuilder(source=self, event_id=event_id_str)
 
     def link_to(self, event_id: str, edge_builder: ProcessStepEdgeBuilder) -> None:
         """Links to the given event ID."""

diff --git a/python/semantic_kernel/processes/process_function_target_builder.py b/python/semantic_kernel/processes/process_function_target_builder.py
@@ -26,18 +26,17 @@ def __init__(
 
         ProcessFunctionTargetBuilder.model_rebuild()
 
-        if isinstance(function_name, Enum):
-            function_name = function_name.value
+        function_name_str: str | None = function_name.value if isinstance(function_name, Enum) else function_name
 
         if isinstance(step, EndStep):
-            function_name = "END"
+            function_name_str = "END"
             parameter_name = None
         else:
-            target = step.resolve_function_target(function_name, parameter_name)
-            function_name = target.function_name
+            target = step.resolve_function_target(function_name_str, parameter_name)
+            function_name_str = target.function_name
             parameter_name = target.parameter_name
 
-        super().__init__(step=step, function_name=function_name, parameter_name=parameter_name)
+        super().__init__(step=step, function_name=function_name_str, parameter_name=parameter_name)
 
     def build(self) -> KernelProcessFunctionTarget:
         """Builds the KernelProcessFunctionTarget."""

diff --git a/python/semantic_kernel/processes/process_step_builder.py b/python/semantic_kernel/processes/process_step_builder.py
@@ -70,19 +70,17 @@ def on_input_event(self, event_id: str | Enum) -> "ProcessStepEdgeBuilder":
         """Creates a new ProcessStepEdgeBuilder for the input event."""
         from semantic_kernel.processes.process_step_edge_builder import ProcessStepEdgeBuilder
 
-        if isinstance(event_id, Enum):
-            event_id = event_id.value
+        event_id_str: str = event_id.value if isinstance(event_id, Enum) else event_id
 
-        return ProcessStepEdgeBuilder(source=self, event_id=event_id)
+        return ProcessStepEdgeBuilder(source=self, event_id=event_id_str)
 
     def on_event(self, event_id: str | Enum) -> "ProcessStepEdgeBuilder":
         """Creates a new ProcessStepEdgeBuilder for the event."""
         from semantic_kernel.processes.process_step_edge_builder import ProcessStepEdgeBuilder
 
-        if isinstance(event_id, Enum):
-            event_id = event_id.value
+        event_id_str: str = event_id.value if isinstance(event_id, Enum) else event_id
 
-        scoped_event_id = self.get_scoped_event_id(event_id)
+        scoped_event_id = self.get_scoped_event_id(event_id_str)
         return ProcessStepEdgeBuilder(source=self, event_id=scoped_event_id)
 
     def resolve_function_target(