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

SIENTIAPDE-1646

Browse Source 
SIENTIAPDE-1646 Update TemporalManager to consistently assign task queues for runtime workflows, enhancing schedule creation logic and ensuring proper handling of runtime names.
This commit is contained in:
vitor-aignosi
2026-05-25 14:46:50 -03:00
parent 7413b70957
commit 7b925069ac
1 changed files with 7 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
orchestrator/activities/temporal_manager.py
View File

@@ -217,6 +217,8 @@ class TemporalManager(SientiaMonitoring):
) )
runtime_name = schedule.get('runtime', 'legacy') if workflow_type in RUNTIME_WORKFLOWS else None runtime_name = schedule.get('runtime', 'legacy') if workflow_type in RUNTIME_WORKFLOWS else None
task_queue_name = build_queue_name(workflow_type, runtime_name)
schedule['task_queue'] = task_queue_name
await client.create_schedule( await client.create_schedule(
schedule_name, schedule_name,
@@ -225,9 +227,7 @@ class TemporalManager(SientiaMonitoring):
workflow_type, workflow_type,
schedule, schedule,
id=schedule_name, id=schedule_name,
task_queue=build_queue_name( task_queue=task_queue_name,
workflow_type, runtime_name
),
execution_timeout=timedelta(seconds=execution_timeout_seconds), execution_timeout=timedelta(seconds=execution_timeout_seconds),
run_timeout=timedelta(seconds=execution_timeout_seconds), run_timeout=timedelta(seconds=execution_timeout_seconds),
task_timeout=timedelta(seconds=task_timeout_seconds), task_timeout=timedelta(seconds=task_timeout_seconds),
@@ -321,6 +321,10 @@ class TemporalManager(SientiaMonitoring):
if not handler: if not handler:
raise ValueError(f'Schedule {schedule_name} not found') raise ValueError(f'Schedule {schedule_name} not found')
workflow_type = schedule['workflow_type']
runtime_name = schedule.get('runtime', 'legacy') if workflow_type in RUNTIME_WORKFLOWS else None
schedule['task_queue'] = build_queue_name(workflow_type, runtime_name)
# fmt: off # fmt: off
async def update_schedule(input_data: ScheduleUpdateInput) -> ScheduleUpdate: # NOSONAR async def update_schedule(input_data: ScheduleUpdateInput) -> ScheduleUpdate: # NOSONAR
schedule_action = input_data.description.schedule.action schedule_action = input_data.description.schedule.action