From 0910c0426afc865c1b0ded1f51b63845eff78102 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 14 Jul 2025 16:21:53 -0300 Subject: [PATCH] SIENTIAPDE-1150 refactor: update orchestration workflow to replace execute_local_activity_method with start_local_activity_method and execute_activity_method with start_activity_method for improved clarity and consistency --- orchestrator/workflows/orchestrator.py | 42 +++++++++++++------------- 1 file changed, 21 insertions(+), 21 deletions(-) diff --git a/orchestrator/workflows/orchestrator.py b/orchestrator/workflows/orchestrator.py index afb24fc..7053746 100644 --- a/orchestrator/workflows/orchestrator.py +++ b/orchestrator/workflows/orchestrator.py @@ -14,7 +14,7 @@ class Orchestrator: input_data['workflow_name'] = 'orchestrator' - pipeline_config_handler = workflow.execute_local_activity_method( + pipeline_config_handler = workflow.start_local_activity_method( Activities.aggregate_documents_in_mongodb, { 'query': input_data['pipelines_query'] @@ -23,7 +23,7 @@ class Orchestrator: start_to_close_timeout=timedelta(seconds=60) ) - opc_servers_handler = workflow.execute_local_activity_method( + opc_servers_handler = workflow.start_local_activity_method( Activities.find_documents_in_mongodb, { 'query': input_data['opc_servers_query'] @@ -32,7 +32,7 @@ class Orchestrator: start_to_close_timeout=timedelta(seconds=60) ) - orchestrated_schedules_handler = workflow.execute_local_activity_method( + orchestrated_schedules_handler = workflow.start_local_activity_method( Activities.find_documents_in_mongodb, { 'query': { @@ -43,13 +43,13 @@ class Orchestrator: start_to_close_timeout=timedelta(seconds=600) ) - current_slot_config_handler = workflow.execute_local_activity_method( + current_slot_config_handler = workflow.start_local_activity_method( Activities.load_opc_slots, retry_policy=retry_policy, start_to_close_timeout=timedelta(seconds=60) ) - active_ingestors_handler = workflow.execute_local_activity_method( + active_ingestors_handler = workflow.start_local_activity_method( Activities.load_active_ingestors, retry_policy=retry_policy, start_to_close_timeout=timedelta(seconds=60) @@ -61,7 +61,7 @@ class Orchestrator: opc_servers = await opc_servers_handler active_ingestors = await active_ingestors_handler - formatted_orchestrated_schedules_handler = workflow.execute_local_activity_method( + formatted_orchestrated_schedules_handler = workflow.start_local_activity_method( Activities.format_schedule_config, { 'schedule_config': orchestrated_schedules @@ -70,7 +70,7 @@ class Orchestrator: start_to_close_timeout=timedelta(seconds=60) ) - schedules_config_handler = workflow.execute_local_activity_method( + schedules_config_handler = workflow.start_local_activity_method( Activities.process_schedules, { 'pipelines': pipeline_config @@ -79,7 +79,7 @@ class Orchestrator: start_to_close_timeout=timedelta(seconds=60) ) - slot_config_handler = workflow.execute_local_activity_method( + slot_config_handler = workflow.start_local_activity_method( Activities.process_slots, { 'opc_servers': opc_servers, @@ -94,7 +94,7 @@ class Orchestrator: slot_config = await slot_config_handler formatted_orchestrated_schedules = await formatted_orchestrated_schedules_handler - schedule_actions_handler = workflow.execute_local_activity_method( + schedule_actions_handler = workflow.start_local_activity_method( Activities.create_schedule_config, { 'current_schedule_config': formatted_orchestrated_schedules, @@ -104,7 +104,7 @@ class Orchestrator: start_to_close_timeout=timedelta(seconds=60) ) - slot_actions_handler = workflow.execute_local_activity_method( + slot_actions_handler = workflow.start_local_activity_method( Activities.create_slot_config, { 'current_slot_config': current_slot_config, @@ -114,7 +114,7 @@ class Orchestrator: start_to_close_timeout=timedelta(seconds=60) ) - normalize_schedules_handler = workflow.execute_local_activity_method( + normalize_schedules_handler = workflow.start_local_activity_method( Activities.normalize_schedules, { 'orchestrated_schedules': formatted_orchestrated_schedules @@ -127,7 +127,7 @@ class Orchestrator: slot_actions = await slot_actions_handler await normalize_schedules_handler - slot_deletion_report_handler = workflow.execute_activity_method( + slot_deletion_report_handler = workflow.start_activity_method( Activities.delete_slots, { 'to_delete': slot_actions['to_delete'] @@ -136,7 +136,7 @@ class Orchestrator: start_to_close_timeout=timedelta(seconds=60) ) - slot_insertion_report_handler = workflow.execute_activity_method( + slot_insertion_report_handler = workflow.start_activity_method( Activities.update_slots, { 'to_insert': slot_actions['to_insert'] @@ -145,7 +145,7 @@ class Orchestrator: start_to_close_timeout=timedelta(seconds=60) ) - schedule_deletion_report_handler = workflow.execute_activity_method( + schedule_deletion_report_handler = workflow.start_activity_method( Activities.delete_schedules, { 'schedules': schedule_actions['to_delete'] @@ -154,7 +154,7 @@ class Orchestrator: start_to_close_timeout=timedelta(seconds=60) ) - schedule_insertion_report_handler = workflow.execute_activity_method( + schedule_insertion_report_handler = workflow.start_activity_method( Activities.create_schedules, { 'schedules': schedule_actions['to_create'] @@ -163,7 +163,7 @@ class Orchestrator: start_to_close_timeout=timedelta(seconds=60) ) - schedule_update_report_handler = workflow.execute_activity_method( + schedule_update_report_handler = workflow.start_activity_method( Activities.update_schedules, { 'schedules': schedule_actions['to_update'] @@ -178,7 +178,7 @@ class Orchestrator: schedule_insertion_report = await schedule_insertion_report_handler schedule_update_report = await schedule_update_report_handler - schedule_report_handler = workflow.execute_activity_method( + schedule_report_handler = workflow.start_activity_method( Activities.report_schedule_orchestration, { 'created_schedules': schedule_insertion_report, @@ -189,7 +189,7 @@ class Orchestrator: start_to_close_timeout=timedelta(seconds=60) ) - slot_report_handler = workflow.execute_activity_method( + slot_report_handler = workflow.start_activity_method( Activities.report_slot_orchestration, { 'inserted_slots': slot_insertion_report, @@ -199,7 +199,7 @@ class Orchestrator: start_to_close_timeout=timedelta(seconds=60) ) - update_pipelines_timestamps_handler = workflow.execute_activity_method( + update_pipelines_timestamps_handler = workflow.start_activity_method( Activities.update_pipelines_timestamps, { 'updated_pipelines': schedule_update_report @@ -208,7 +208,7 @@ class Orchestrator: start_to_close_timeout=timedelta(seconds=60) ) - create_pipelines_timestamps_handler = workflow.execute_activity_method( + create_pipelines_timestamps_handler = workflow.start_activity_method( Activities.create_pipelines_timestamps, { 'created_pipelines': schedule_insertion_report @@ -217,7 +217,7 @@ class Orchestrator: start_to_close_timeout=timedelta(seconds=60) ) - delete_pipelines_timestamps_handler = workflow.execute_activity_method( + delete_pipelines_timestamps_handler = workflow.start_activity_method( Activities.delete_pipelines_timestamps, { 'deleted_pipelines': schedule_deletion_report