aa1850ce47
Refactor camelCase conversion in prepare_worker.py to snake_case for consistency in queue naming. Updated function name and logic accordingly.
73 lines
2.7 KiB
Python
73 lines
2.7 KiB
Python
import os
|
|
import re
|
|
from collections.abc import Sequence
|
|
from typing import Any
|
|
|
|
from sientia_do.observability.logger import Logger
|
|
from temporalio.client import Client
|
|
from temporalio.worker import PollerBehaviorAutoscaling, Worker
|
|
|
|
parameters = [
|
|
('MAX_CONCURRENT_WORKFLOW_TASKS', '200'),
|
|
('MAX_CONCURRENT_ACTIVITIES', '200'),
|
|
('MAX_CONCURRENT_LOCAL_ACTIVITIES', '200'),
|
|
('MAX_CACHED_WORKFLOWS', '200'),
|
|
('WORKFLOW_POLLER_BEHAVIUR_MINIMUM', '10'),
|
|
('WORKFLOW_POLLER_BEHAVIUR_INITIAL', '100'),
|
|
('WORKFLOW_POLLER_BEHAVIUR_MAXIMUM', '200'),
|
|
('ACTIVITY_POLLER_BEHAVIUR_MINIMUM', '10'),
|
|
('ACTIVITY_POLLER_BEHAVIUR_INITIAL', '100'),
|
|
('ACTIVITY_POLLER_BEHAVIUR_MAXIMUM', '200'),
|
|
]
|
|
|
|
|
|
def camel_to_snake(text: str) -> str:
|
|
"""Convert camelCase or PascalCase to snake_case."""
|
|
text = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', text)
|
|
text = re.sub('([a-z0-9])([A-Z])', r'\1_\2', text)
|
|
return text.lower()
|
|
|
|
|
|
def prepare_worker(
|
|
main_workflow: type,
|
|
other_workflows: Sequence[type],
|
|
activities: Sequence[Any],
|
|
temporal_client: Client,
|
|
logger: Logger,
|
|
) -> Worker:
|
|
main_workflow_name = main_workflow.__name__.upper()
|
|
|
|
queue_name = f'{camel_to_snake(main_workflow.__name__)}_queue'
|
|
|
|
local_workflow_parameters = {}
|
|
|
|
for parameter in parameters:
|
|
local_workflow_parameters[parameter[0]] = int(
|
|
os.getenv(main_workflow_name + '_' + parameter[0], parameter[1])
|
|
)
|
|
|
|
logger.info(f'Preparing worker for {main_workflow_name} with queue {queue_name}')
|
|
|
|
return Worker(
|
|
temporal_client,
|
|
task_queue=queue_name,
|
|
workflows=[main_workflow, *other_workflows],
|
|
activities=[*activities],
|
|
max_concurrent_workflow_tasks=local_workflow_parameters['MAX_CONCURRENT_WORKFLOW_TASKS'],
|
|
max_concurrent_activities=local_workflow_parameters['MAX_CONCURRENT_ACTIVITIES'],
|
|
max_concurrent_local_activities=local_workflow_parameters[
|
|
'MAX_CONCURRENT_LOCAL_ACTIVITIES'
|
|
],
|
|
max_cached_workflows=local_workflow_parameters['MAX_CACHED_WORKFLOWS'],
|
|
workflow_task_poller_behavior=PollerBehaviorAutoscaling(
|
|
minimum=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIUR_MINIMUM'],
|
|
initial=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIUR_INITIAL'],
|
|
maximum=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIUR_MAXIMUM'],
|
|
),
|
|
activity_task_poller_behavior=PollerBehaviorAutoscaling(
|
|
minimum=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIUR_MINIMUM'],
|
|
initial=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIUR_INITIAL'],
|
|
maximum=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIUR_MAXIMUM'],
|
|
),
|
|
)
|