Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion backend/api_v2/deployment_helper.py
Original file line number Diff line number Diff line change
Expand Up @@ -191,7 +191,7 @@ def execute_workflow(
include_metrics: bool = False,
include_extracted_text: bool = False,
use_file_history: bool = False,
tag_names: list[str] = [],
tag_names: list[str] | None = None,
llm_profile_id: str | None = None,
hitl_queue_name: str | None = None,
hitl_packet_id: str | None = None,
Expand Down Expand Up @@ -222,6 +222,7 @@ def execute_workflow(
Rate limiting is handled at the view layer. This method should be called
after rate limit checks have passed, with a pre-acquired execution_id.
"""
tag_names = tag_names or []
# Use provided execution_id or generate one (for backward compatibility)
if execution_id is None:
execution_id = str(uuid.uuid4())
Expand Down
6 changes: 4 additions & 2 deletions backend/workflow_manager/endpoint_v2/queue_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,12 +25,13 @@ class QueueUtils:
_hitl_connectors = {}

@staticmethod
def get_queue_inst(connector_settings: dict[str, Any] = {}) -> UnstractQueue:
def get_queue_inst(connector_settings: dict[str, Any] | None = None) -> UnstractQueue:
"""Get queue connector instance based on configuration.

For HITL operations, this can return PostgreSQL, Hybrid, or Redis connectors
based on the HITL_QUEUE_BACKEND setting.
"""
connector_settings = connector_settings or {}
# Check if caller explicitly wants the default (non-HITL) backend
# This is used by HybridQueue to get the actual Redis connector without recursion
force_default = connector_settings.get("force_default_backend", False)
Expand Down Expand Up @@ -73,7 +74,7 @@ def get_queue_inst(connector_settings: dict[str, Any] = {}) -> UnstractQueue:

@staticmethod
def get_hitl_queue_inst(
backend: str, connector_settings: dict[str, Any] = {}
backend: str, connector_settings: dict[str, Any] | None = None
) -> UnstractQueue:
"""Get HITL-specific queue connector instance with dynamic imports.

Expand All @@ -91,6 +92,7 @@ def get_hitl_queue_inst(
Raises:
UnstractQueueException: When HITL connectors are not available
"""
connector_settings = connector_settings or {}
# For Redis backend, use default connector
if backend == "redis":
# Strip HITL flag to force default (non-HITL) connector path
Expand Down
3 changes: 2 additions & 1 deletion backend/workflow_manager/endpoint_v2/source.py
Original file line number Diff line number Diff line change
Expand Up @@ -912,7 +912,7 @@ def _create_file_hash(
)

def list_files_from_source(
self, file_hashes: dict[str, FileHash] = {}
self, file_hashes: dict[str, FileHash] | None = None
) -> tuple[dict[str, FileHash], int]:
"""List files from source connector.

Expand All @@ -922,6 +922,7 @@ def list_files_from_source(
tuple[dict[str, FileHash], int]: A dictionary of FileHashes,
along with the total count of matched files.
"""
file_hashes = file_hashes or {}
connection_type = self.endpoint.connection_type
if connection_type == WorkflowEndpoint.ConnectionType.FILESYSTEM:
files = self.list_files_from_file_connector()
Expand Down
5 changes: 4 additions & 1 deletion backend/workflow_manager/workflow_v2/dto.py
Original file line number Diff line number Diff line change
Expand Up @@ -87,14 +87,17 @@ def _remove_item_top_metadata(item: dict, keys_to_remove: list[str]) -> None:
else:
item.pop("metadata", None)

def remove_result_metadata_keys(self, keys_to_remove: list[str] = []) -> None:
def remove_result_metadata_keys(
self, keys_to_remove: list[str] | None = None
) -> None:
"""Removes specified keys from the 'metadata' dictionary within each
'result' dictionary in the 'result' list attribute of the instance. If
'keys_to_remove' is empty, the 'metadata' key itself is removed.

Args:
keys_to_remove (List[str]): List of keys to be removed from 'metadata'.
"""
keys_to_remove = keys_to_remove or []
if not isinstance(self.result, list):
return

Expand Down
3 changes: 2 additions & 1 deletion backend/workflow_manager/workflow_v2/views.py
Original file line number Diff line number Diff line change
Expand Up @@ -302,9 +302,10 @@ def execute_workflow(
execution_action: str | None = None,
execution_id: str | None = None,
pipeline_guid: str | None = None,
hash_values_of_files: dict[str, FileHash] = {},
hash_values_of_files: dict[str, FileHash] | None = None,
use_file_history: bool = False,
) -> ExecutionResponse:
hash_values_of_files = hash_values_of_files or {}
# Detect if this is an API execution by checking connector types
is_api_execution = WorkflowEndpointUtils.is_api_workflow(workflow)

Expand Down
9 changes: 6 additions & 3 deletions backend/workflow_manager/workflow_v2/workflow_helper.py
Original file line number Diff line number Diff line change
Expand Up @@ -302,7 +302,7 @@
def run_workflow(
workflow: Workflow,
workflow_execution: WorkflowExecution,
hash_values_of_files: dict[str, FileHash] = {},
hash_values_of_files: dict[str, FileHash] | None = None,
organization_id: str | None = None,
pipeline_id: str | None = None,
scheduled: bool = False,
Expand All @@ -314,6 +314,7 @@
packet_id: str | None = None,
custom_data: dict[str, Any] | None = None,
) -> ExecutionResponse:
hash_values_of_files = hash_values_of_files or {}
tool_instances: list[ToolInstance] = (
ToolInstanceHelper.get_tool_instances_by_workflow(
workflow.id, ToolInstanceKey.STEP
Expand Down Expand Up @@ -875,16 +876,17 @@
return execution_response.result

@staticmethod
def complete_execution(

Check failure on line 879 in backend/workflow_manager/workflow_v2/workflow_helper.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Refactor this function to reduce its Cognitive Complexity from 16 to the 15 allowed.

See more on https://sonarcloud.io/project/issues?id=Zipstack_unstract&issues=AaAjvol-zEBYCxsVdT5X&open=AaAjvol-zEBYCxsVdT5X&pullRequest=2252
workflow: Workflow,
execution_id: str | None = None,
pipeline_id: str | None = None,
execution_mode: WorkflowExecution | None = WorkflowExecution.Mode.QUEUE,
hash_values_of_files: dict[str, FileHash] = {},
hash_values_of_files: dict[str, FileHash] | None = None,
use_file_history: bool = False,
timeout: int | None = None,
is_api_execution: bool = False,
) -> ExecutionResponse:
hash_values_of_files = hash_values_of_files or {}
if pipeline_id:
logger.info(f"Executing pipeline: {pipeline_id}")
# Create a new WorkflowExecution entity for each pipeline execution.
Expand Down Expand Up @@ -1020,12 +1022,13 @@
raise WorkflowExecutionNotExist()

@staticmethod
def step_execution(

Check failure on line 1025 in backend/workflow_manager/workflow_v2/workflow_helper.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Refactor this function to reduce its Cognitive Complexity from 16 to the 15 allowed.

See more on https://sonarcloud.io/project/issues?id=Zipstack_unstract&issues=AaAjvol-zEBYCxsVdT5Y&open=AaAjvol-zEBYCxsVdT5Y&pullRequest=2252
workflow: Workflow,
execution_action: str,
execution_id: str | None = None,
hash_values_of_files: dict[str, FileHash] = {},
hash_values_of_files: dict[str, FileHash] | None = None,
) -> ExecutionResponse:
hash_values_of_files = hash_values_of_files or {}
if execution_action is Workflow.ExecutionAction.START.value: # type: ignore
if execution_id is None:
return WorkflowHelper.create_and_make_execution_response(
Expand Down