From ba87c489fe77485ce53e1836d5c898991e372861 Mon Sep 17 00:00:00 2001 From: Evan Mattson Date: Thu, 14 Nov 2024 15:15:17 -0500 Subject: [PATCH] Fix mypy errors --- .../dapr_runtime/dapr_kernel_process.py | 11 ++++++----- .../local_runtime/local_kernel_process.py | 11 ++++++----- .../semantic_kernel/processes/process_builder.py | 16 +++++++--------- .../processes/process_function_target_builder.py | 11 +++++------ .../processes/process_step_builder.py | 10 ++++------ 5 files changed, 28 insertions(+), 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 index a54458e5e7a9..cd43c67956e9 100644 --- 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 index 19d61a20cc27..6a7408b5b29b 100644 --- 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 index 7628185a91c9..c4e024298af3 100644 --- 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 index 7d2339990fc4..30a695273721 100644 --- 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 index 99e1a7283fb6..ee09eb5f92dd 100644 --- 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(