Worker properties config (#331)

Author: IgorChvyrov-sm

src/conductor/asyncio_client/automator/task_runner.py

@@ -243,21 +243,19 @@ def __set_worker_properties(self) -> None:
         polling_interval = self.__get_property_value_from_env(
             "polling_interval", task_type
         )
+
         if polling_interval:
             try:
                 self.worker.poll_interval = float(polling_interval)
-            except Exception:
+            except Exception as e:
                 logger.error(
-                    "Error converting polling_interval to float value: %s",
+                    "Error converting polling_interval to float value: %s, exception: %s",
                     polling_interval,
+                    e,
                 )
                 self.worker.poll_interval = (
                     self.worker.get_polling_interval_in_seconds()
                 )
-        else:
-            logger.error(
-                "Exception in reading polling_interval from environment variable",
-            )
 
     def __get_property_value_from_env(self, prop, task_type):
         """

src/conductor/asyncio_client/configuration/configuration.py

@@ -26,8 +26,9 @@ class Configuration:
 
     Worker Properties (via environment variables):
     ----------------------------------------------
-    CONDUCTOR_WORKER_POLLING_INTERVAL: Default polling interval in seconds
     CONDUCTOR_WORKER_DOMAIN: Default worker domain
+    CONDUCTOR_WORKER_POLL_INTERVAL: Polling interval in milliseconds (default: 100)
+    CONDUCTOR_WORKER_POLL_INTERVAL_SECONDS: Polling interval in seconds (default: 0)
     CONDUCTOR_WORKER_<TASK_TYPE>_POLLING_INTERVAL: Task-specific polling interval
     CONDUCTOR_WORKER_<TASK_TYPE>_DOMAIN: Task-specific domain
 
@@ -57,8 +58,9 @@ def __init__(
         auth_secret: Optional[str] = None,
         debug: bool = False,
         # Worker properties
-        default_polling_interval: Optional[float] = None,
-        default_domain: Optional[str] = None,
+        polling_interval: Optional[int] = None,
+        domain: Optional[str] = None,
+        polling_interval_seconds: Optional[int] = None,
         # HTTP Configuration parameters
         api_key: Optional[Dict[str, str]] = None,
         api_key_prefix: Optional[Dict[str, str]] = None,
@@ -88,10 +90,12 @@ def __init__(
             Authentication key secret. If not provided, reads from CONDUCTOR_AUTH_SECRET env var.
         debug : bool, optional
             Enable debug logging. Default is False.
-        default_polling_interval : float, optional
-            Default polling interval for workers in seconds.
-        default_domain : str, optional
-            Default domain for workers.
+        polling_interval : int, optional
+            Polling interval in milliseconds. If not provided, reads from CONDUCTOR_WORKER_POLL_INTERVAL env var.
+        domain : str, optional
+            Worker domain. If not provided, reads from CONDUCTOR_WORKER_DOMAIN env var.
+        polling_interval_seconds : int, optional
+            Polling interval in seconds. If not provided, reads from CONDUCTOR_WORKER_POLL_INTERVAL_SECONDS env var.
         **kwargs : Any
             Additional parameters passed to HttpConfiguration.
         """
@@ -116,11 +120,14 @@ def __init__(
         else:
             self.auth_secret = os.getenv("CONDUCTOR_AUTH_SECRET")
 
-        # Worker properties with environment variable fallback
-        self.default_polling_interval = default_polling_interval or self._get_env_float(
-            "CONDUCTOR_WORKER_POLLING_INTERVAL", 1.0
+        # Additional worker properties with environment variable fallback
+        self.polling_interval = polling_interval or self._get_env_int(
+            "CONDUCTOR_WORKER_POLL_INTERVAL", 100
+        )
+        self.domain = domain or os.getenv("CONDUCTOR_WORKER_DOMAIN", "default_domain")
+        self.polling_interval_seconds = polling_interval_seconds or self._get_env_int(
+            "CONDUCTOR_WORKER_POLL_INTERVAL_SECONDS", 0
         )
-        self.default_domain = default_domain or os.getenv("CONDUCTOR_WORKER_DOMAIN")
 
         # Store additional worker properties
         self._worker_properties: Dict[str, Dict[str, Any]] = {}
@@ -234,10 +241,12 @@ def get_worker_property_value(
             return self._convert_property_value(property_name, value)
 
         # Return default value
-        if property_name == "polling_interval":
-            return self.default_polling_interval
         elif property_name == "domain":
-            return self.default_domain
+            return self.domain
+        elif property_name == "polling_interval":
+            return self.polling_interval
+        elif property_name == "poll_interval_seconds":
+            return self.polling_interval_seconds
 
         return None
 
@@ -248,7 +257,13 @@ def _convert_property_value(self, property_name: str, value: str) -> Any:
                 return float(value)
             except (ValueError, TypeError):
                 self.logger.warning("Invalid polling_interval value: %s", value)
-                return self.default_polling_interval
+                return self.polling_interval
+        elif property_name == "polling_interval_seconds":
+            try:
+                return float(value)
+            except (ValueError, TypeError):
+                self.logger.warning("Invalid polling_interval_seconds value: %s", value)
+                return self.polling_interval_seconds
 
         # For other properties, return as string
         return value
@@ -325,6 +340,37 @@ def get_domain(self, task_type: Optional[str] = None) -> Optional[str]:
         """
         return self.get_worker_property_value("domain", task_type)
 
+    def get_poll_interval(self, task_type: Optional[str] = None) -> int:
+        """
+        Get polling interval in milliseconds for a task type with environment variable support.
+
+        Parameters:
+        -----------
+        task_type : str, optional
+            Task type for task-specific configuration
+
+        Returns:
+        --------
+        int
+            Polling interval in milliseconds
+        """
+        if task_type:
+            value = self.get_worker_property_value("polling_interval", task_type)
+            if value is not None:
+                return int(value)
+        return self.polling_interval
+
+    def get_poll_interval_seconds(self) -> int:
+        """
+        Get polling interval in seconds.
+
+        Returns:
+        --------
+        int
+            Polling interval in seconds
+        """
+        return self.polling_interval_seconds
+
     # Properties for commonly used HTTP configuration attributes
     @property
     def host(self) -> str:

src/conductor/asyncio_client/worker/worker.py

@@ -9,14 +9,15 @@
 from typing import Any, Callable, Optional, Union
 
 from conductor.asyncio_client.adapters.models.task_adapter import TaskAdapter
-from conductor.asyncio_client.adapters.models.task_exec_log_adapter import \
-    TaskExecLogAdapter
-from conductor.asyncio_client.adapters.models.task_result_adapter import \
-    TaskResultAdapter
+from conductor.asyncio_client.adapters.models.task_exec_log_adapter import (
+    TaskExecLogAdapter,
+)
+from conductor.asyncio_client.adapters.models.task_result_adapter import (
+    TaskResultAdapter,
+)
 from conductor.asyncio_client.configuration import Configuration
 from conductor.asyncio_client.adapters import ApiClient
-from conductor.asyncio_client.worker.worker_interface import (
-    DEFAULT_POLLING_INTERVAL, WorkerInterface)
+from conductor.asyncio_client.worker.worker_interface import WorkerInterface
 from conductor.shared.automator import utils
 from conductor.shared.automator.utils import convert_from_dict_or_list
 from conductor.shared.http.enums import TaskResultStatus
@@ -60,15 +61,10 @@ def __init__(
     ):
         super().__init__(task_definition_name)
         self.api_client = ApiClient()
-        if poll_interval is None:
-            self.poll_interval = DEFAULT_POLLING_INTERVAL
-        else:
-            self.poll_interval = deepcopy(poll_interval)
-        self.domain = deepcopy(domain)
-        if worker_id is None:
-            self.worker_id = deepcopy(super().get_identity())
-        else:
-            self.worker_id = deepcopy(worker_id)
+        self.config = Configuration()
+        self.poll_interval = poll_interval or self.config.get_poll_interval()
+        self.domain = domain or self.config.get_domain()
+        self.worker_id = worker_id or super().get_identity()
         self.execute_function = deepcopy(execute_function)
 
     def execute(self, task: TaskAdapter) -> TaskResultAdapter:

src/conductor/asyncio_client/worker/worker_task.py

@@ -3,8 +3,8 @@
 import functools
 from typing import Optional
 
-from conductor.asyncio_client.automator.task_handler import \
-    register_decorated_fn
+from conductor.asyncio_client.automator.task_handler import register_decorated_fn
+from conductor.asyncio_client.configuration.configuration import Configuration
 from conductor.asyncio_client.workflow.task.simple_task import SimpleTask
 
 
@@ -15,12 +15,17 @@ def WorkerTask(
     worker_id: Optional[str] = None,
     poll_interval_seconds: int = 0,
 ):
+    config = Configuration()
+
+    poll_interval = poll_interval or config.get_poll_interval()
+    domain = domain or config.get_domain()
+    poll_interval_seconds = poll_interval_seconds or config.get_poll_interval_seconds()
+
     poll_interval_millis = poll_interval
     if poll_interval_seconds > 0:
         poll_interval_millis = 1000 * poll_interval_seconds
 
     def worker_task_func(func):
-
         register_decorated_fn(
             name=task_definition_name,
             poll_interval=poll_interval_millis,
@@ -52,6 +57,11 @@ def worker_task(
     domain: Optional[str] = None,
     worker_id: Optional[str] = None,
 ):
+    config = Configuration()
+
+    poll_interval_millis = poll_interval_millis or config.get_poll_interval()
+    domain = domain or config.get_domain()
+
     def worker_task_func(func):
         register_decorated_fn(
             name=task_definition_name,

src/conductor/client/automator/task_runner.py

@@ -244,27 +244,20 @@ def __set_worker_properties(self) -> None:
         polling_interval = self.__get_property_value_from_env(
             "polling_interval", task_type
         )
+
         if polling_interval:
             try:
                 self.worker.poll_interval = float(polling_interval)
-            except Exception:
+            except Exception as e:
                 logger.error(
-                    "Error converting polling_interval to float value: %s",
+                    "Error converting polling_interval to float value: %s, exception: %s",
                     polling_interval,
+                    e,
                 )
                 self.worker.poll_interval = (
                     self.worker.get_polling_interval_in_seconds()
                 )
 
-        if polling_interval:
-            try:
-                self.worker.poll_interval = float(polling_interval)
-            except Exception as e:
-                logger.error(
-                    "Exception in reading polling interval from environment variable: %s",
-                    e,
-                )
-
     def __get_property_value_from_env(self, prop, task_type):
         """
         get the property from the env variable

src/conductor/client/configuration/configuration.py

@@ -20,6 +20,9 @@ def __init__(
         authentication_settings: AuthenticationSettings = None,
         server_api_url: Optional[str] = None,
         auth_token_ttl_min: int = 45,
+        polling_interval: Optional[float] = None,
+        domain: Optional[str] = None,
+        polling_interval_seconds: Optional[float] = None,
     ):
         if server_api_url is not None:
             self.host = server_api_url
@@ -79,6 +82,15 @@ def __init__(
         self.token_update_time = 0
         self.auth_token_ttl_msec = auth_token_ttl_min * 60 * 1000
 
+        # Worker properties
+        self.polling_interval = polling_interval or self._get_env_float(
+            "CONDUCTOR_WORKER_POLL_INTERVAL", 100
+        )
+        self.domain = domain or os.getenv("CONDUCTOR_WORKER_DOMAIN", "default_domain")
+        self.polling_interval_seconds = polling_interval_seconds or self._get_env_float(
+            "CONDUCTOR_WORKER_POLL_INTERVAL_SECONDS", 0
+        )
+
     @property
     def debug(self):
         """Debug status
@@ -162,3 +174,22 @@ def get_logging_formatted_name(name):
     def update_token(self, token: str) -> None:
         self.AUTH_TOKEN = token
         self.token_update_time = round(time.time() * 1000)
+
+    def _get_env_float(self, env_var: str, default: float) -> float:
+        """Get float value from environment variable with default fallback."""
+        try:
+            value = os.getenv(env_var)
+            if value is not None:
+                return float(value)
+        except (ValueError, TypeError):
+            pass
+        return default
+
+    def get_poll_interval_seconds(self):
+        return self.polling_interval_seconds
+
+    def get_poll_interval(self):
+        return self.polling_interval
+
+    def get_domain(self):
+        return self.domain