gitea_admin/sientia-dataops-orchestrator_temporal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SIENTIAPDE-1150

Browse Source 
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
This commit is contained in:
vitor-aignosi
2025-07-14 16:21:53 -03:00
parent 7f162a7d4e
commit 0910c0426a
1 changed files with 21 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
orchestrator/workflows/orchestrator.py
View File

@@ -14,7 +14,7 @@ class Orchestrator:
input_data['workflow_name'] = '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, Activities.aggregate_documents_in_mongodb,
{ {
'query': input_data['pipelines_query'] 'query': input_data['pipelines_query']
@@ -23,7 +23,7 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=60) 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, Activities.find_documents_in_mongodb,
{ {
'query': input_data['opc_servers_query'] 'query': input_data['opc_servers_query']
@@ -32,7 +32,7 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=60) 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, Activities.find_documents_in_mongodb,
{ {
'query': { 'query': {
@@ -43,13 +43,13 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=600) 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, Activities.load_opc_slots,
retry_policy=retry_policy, retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60) 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, Activities.load_active_ingestors,
retry_policy=retry_policy, retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60) start_to_close_timeout=timedelta(seconds=60)
@@ -61,7 +61,7 @@ class Orchestrator:
opc_servers = await opc_servers_handler opc_servers = await opc_servers_handler
active_ingestors = await active_ingestors_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, Activities.format_schedule_config,
{ {
'schedule_config': orchestrated_schedules 'schedule_config': orchestrated_schedules
@@ -70,7 +70,7 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=60) 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, Activities.process_schedules,
{ {
'pipelines': pipeline_config 'pipelines': pipeline_config
@@ -79,7 +79,7 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=60) 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, Activities.process_slots,
{ {
'opc_servers': opc_servers, 'opc_servers': opc_servers,
@@ -94,7 +94,7 @@ class Orchestrator:
slot_config = await slot_config_handler slot_config = await slot_config_handler
formatted_orchestrated_schedules = await formatted_orchestrated_schedules_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, Activities.create_schedule_config,
{ {
'current_schedule_config': formatted_orchestrated_schedules, 'current_schedule_config': formatted_orchestrated_schedules,
@@ -104,7 +104,7 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=60) 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, Activities.create_slot_config,
{ {
'current_slot_config': current_slot_config, 'current_slot_config': current_slot_config,
@@ -114,7 +114,7 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=60) 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, Activities.normalize_schedules,
{ {
'orchestrated_schedules': formatted_orchestrated_schedules 'orchestrated_schedules': formatted_orchestrated_schedules
@@ -127,7 +127,7 @@ class Orchestrator:
slot_actions = await slot_actions_handler slot_actions = await slot_actions_handler
await normalize_schedules_handler await normalize_schedules_handler
slot_deletion_report_handler = workflow.execute_activity_method( slot_deletion_report_handler = workflow.start_activity_method(
Activities.delete_slots, Activities.delete_slots,
{ {
'to_delete': slot_actions['to_delete'] 'to_delete': slot_actions['to_delete']
@@ -136,7 +136,7 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=60) 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, Activities.update_slots,
{ {
'to_insert': slot_actions['to_insert'] 'to_insert': slot_actions['to_insert']
@@ -145,7 +145,7 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=60) 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, Activities.delete_schedules,
{ {
'schedules': schedule_actions['to_delete'] 'schedules': schedule_actions['to_delete']
@@ -154,7 +154,7 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=60) 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, Activities.create_schedules,
{ {
'schedules': schedule_actions['to_create'] 'schedules': schedule_actions['to_create']
@@ -163,7 +163,7 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=60) 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, Activities.update_schedules,
{ {
'schedules': schedule_actions['to_update'] 'schedules': schedule_actions['to_update']
@@ -178,7 +178,7 @@ class Orchestrator:
schedule_insertion_report = await schedule_insertion_report_handler schedule_insertion_report = await schedule_insertion_report_handler
schedule_update_report = await schedule_update_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, Activities.report_schedule_orchestration,
{ {
'created_schedules': schedule_insertion_report, 'created_schedules': schedule_insertion_report,
@@ -189,7 +189,7 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=60) 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, Activities.report_slot_orchestration,
{ {
'inserted_slots': slot_insertion_report, 'inserted_slots': slot_insertion_report,
@@ -199,7 +199,7 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=60) 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, Activities.update_pipelines_timestamps,
{ {
'updated_pipelines': schedule_update_report 'updated_pipelines': schedule_update_report
@@ -208,7 +208,7 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=60) 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, Activities.create_pipelines_timestamps,
{ {
'created_pipelines': schedule_insertion_report 'created_pipelines': schedule_insertion_report
@@ -217,7 +217,7 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=60) 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, Activities.delete_pipelines_timestamps,
{ {
'deleted_pipelines': schedule_deletion_report 'deleted_pipelines': schedule_deletion_report