fix: Removed assert on task ID NULL, (#1147)

fix: Removed assert on task ID NULL, avoided calling shared_task on scheduled run
Zipstack · Feb 22, 2025 · f96dedc · f96dedc
1 parent 1640f4f
commit f96dedc
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 7 deletions.
diff --git a/backend/workflow_manager/workflow_v2/execution.py b/backend/workflow_manager/workflow_v2/execution.py
@@ -415,9 +415,11 @@ def update_execution_err(execution_id: str, err_msg: str = "") -> WorkflowExecut
     @staticmethod
     def update_execution_task(execution_id: str, task_id: str) -> None:
         try:
-            assert (
-                task_id is not None
-            ), f"task_id is NULL for execution_id: {execution_id}"
+            if not task_id:
+                logger.warning(
+                    f"task_id: '{task_id}' is NULL / empty for "
+                    f"execution_id: {execution_id}, expected to have a UUID"
+                )
             execution = WorkflowExecution.objects.get(pk=execution_id)
             # TODO: Review if status should be updated to EXECUTING
             execution.task_id = task_id

diff --git a/backend/workflow_manager/workflow_v2/workflow_helper.py b/backend/workflow_manager/workflow_v2/workflow_helper.py
@@ -752,18 +752,21 @@ def complete_execution(
                 )
                 return response
             else:
-                execution_result = WorkflowHelper.execute_bin(
-                    schema_name=org_schema,
+                task_id = current_task.request.id
+                # TODO: Remove this if scheduled runs work
+                StateStore.set(Account.ORGANIZATION_ID, org_schema)
+                execution_result = WorkflowHelper.execute_workflow(
+                    organization_id=org_schema,
+                    task_id=task_id,
                     workflow_id=workflow.id,
                     execution_id=workflow_execution.id,
                     hash_values_of_files=hash_values_of_files,
                     scheduled=True,
                     execution_mode=execution_mode,
                     pipeline_id=pipeline_id,
-                    log_events_id=log_events_id,
                     use_file_history=use_file_history,
+                    log_events_id=log_events_id,
                 )
-
             updated_execution = WorkflowExecution.objects.get(id=execution_id)
             execution_response = ExecutionResponse(
                 workflow.id,