Added worker configuration

Author: IgorChvyrov-sm

src/conductor/asyncio_client/automator/task_runner.py

@@ -240,24 +240,28 @@ def __set_worker_properties(self) -> None:
         else:
             self.worker.domain = self.worker.get_domain()
 
+        # Try both naming conventions for polling interval
         polling_interval = self.__get_property_value_from_env(
-            "polling_interval", task_type
+            "poll_interval", task_type
         )
+        if not polling_interval:
+            # Fallback to the "polling_interval" naming convention
+            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()
+                    self.worker.get_polling_interval_in_seconds() * 1000
                 )
-        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,
+        poll_interval: Optional[int] = None,
+        domain: Optional[str] = None,
+        poll_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.
+        poll_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.
+        poll_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,10 @@ 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
-        )
-        self.default_domain = default_domain or os.getenv("CONDUCTOR_WORKER_DOMAIN")
+        # Additional worker properties with environment variable fallback
+        self.poll_interval = poll_interval or self._get_env_int("CONDUCTOR_WORKER_POLL_INTERVAL", 100)
+        self.domain = domain or os.getenv("CONDUCTOR_WORKER_DOMAIN", "default_domain")
+        self.poll_interval_seconds = poll_interval_seconds or self._get_env_int("CONDUCTOR_WORKER_POLL_INTERVAL_SECONDS", 0)
 
         # Store additional worker properties
         self._worker_properties: Dict[str, Dict[str, Any]] = {}
@@ -234,10 +237,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 == "poll_interval":
+            return self.poll_interval
+        elif property_name == "poll_interval_seconds":
+            return self.poll_interval_seconds
 
         return None
 
@@ -247,8 +252,14 @@ def _convert_property_value(self, property_name: str, value: str) -> Any:
             try:
                 return float(value)
             except (ValueError, TypeError):
-                self.logger.warning("Invalid polling_interval value: %s", value)
-                return self.default_polling_interval
+                self.logger.warning("Invalid poll_interval value: %s", value)
+                return self.poll_interval
+        elif property_name == "poll_interval_seconds":
+            try:
+                return float(value)
+            except (ValueError, TypeError):
+                self.logger.warning("Invalid poll_interval_seconds value: %s", value)
+                return self.poll_interval_seconds
 
         # For other properties, return as string
         return value
@@ -325,6 +336,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("poll_interval", task_type)
+            if value is not None:
+                return int(value)
+        return self.poll_interval
+
+    def get_poll_interval_seconds(self) -> int:
+        """
+        Get polling interval in seconds.
+
+        Returns:
+        --------
+        int
+            Polling interval in seconds
+        """
+        return self.poll_interval_seconds
+
     # Properties for commonly used HTTP configuration attributes
     @property
     def host(self) -> str:

src/conductor/asyncio_client/worker/worker.py

@@ -15,8 +15,7 @@
     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 +59,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,24 +3,30 @@
 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
 
 
 def WorkerTask(
     task_definition_name: str,
     poll_interval: int = 100,
-    domain: Optional[str] = None,
-    worker_id: Optional[str] = None,
+    domain: str = None,
+    worker_id: 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()
+
+    # Calculate poll interval in milliseconds
     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 +58,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

@@ -241,28 +241,27 @@ def __set_worker_properties(self) -> None:
         else:
             self.worker.domain = self.worker.get_domain()
 
+        # Try both naming conventions for polling interval
         polling_interval = self.__get_property_value_from_env(
-            "polling_interval", task_type
+            "poll_interval", task_type
         )
+        if not polling_interval:
+            # Fallback to the "polling_interval" naming convention
+            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,
+                    self.worker.get_polling_interval_in_seconds() * 1000
                 )
 
     def __get_property_value_from_env(self, prop, task_type):

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,
+        poll_interval: Optional[float] = None,
+        domain: Optional[str] = None,
+        poll_interval_seconds: Optional[float] = None,
     ):
         if server_api_url is not None:
             self.host = server_api_url
@@ -79,6 +82,11 @@ def __init__(
         self.token_update_time = 0
         self.auth_token_ttl_msec = auth_token_ttl_min * 60 * 1000
 
+        # Worker properties
+        self.poll_interval = poll_interval or self._get_env_float("CONDUCTOR_WORKER_POLL_INTERVAL", 100)
+        self.domain = domain or os.getenv("CONDUCTOR_WORKER_DOMAIN", "default_domain")
+        self.poll_interval_seconds = poll_interval_seconds or self._get_env_float("CONDUCTOR_WORKER_POLL_INTERVAL_SECONDS", 0)
+
     @property
     def debug(self):
         """Debug status
@@ -162,3 +170,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.poll_interval_seconds
+
+    def get_poll_interval(self):
+        return self.poll_interval
+
+    def get_domain(self):
+        return self.domain

src/conductor/client/worker/worker.py

@@ -18,7 +18,7 @@
 from conductor.client.http.models.task_result import TaskResult
 from conductor.shared.http.enums import TaskResultStatus
 from conductor.shared.worker.exception import NonRetryableException
-from conductor.client.worker.worker_interface import WorkerInterface, DEFAULT_POLLING_INTERVAL
+from conductor.client.worker.worker_interface import WorkerInterface
 
 ExecuteTaskFunction = Callable[
     [
@@ -57,11 +57,16 @@ def __init__(self,
                  ) -> Self:
         super().__init__(task_definition_name)
         self.api_client = ApiClient()
+        self.config = Configuration()
+
         if poll_interval is None:
-            self.poll_interval = DEFAULT_POLLING_INTERVAL
+            self.poll_interval = self.config.get_poll_interval()
         else:
             self.poll_interval = deepcopy(poll_interval)
-        self.domain = deepcopy(domain)
+        if domain is None:
+            self.domain = self.config.get_domain()
+        else:
+            self.domain = deepcopy(domain)
         if worker_id is None:
             self.worker_id = deepcopy(super().get_identity())
         else:

src/conductor/client/worker/worker_interface.py

@@ -5,6 +5,7 @@
 
 from conductor.client.http.models.task import Task
 from conductor.client.http.models.task_result import TaskResult
+from conductor.client.configuration.configuration import Configuration
 
 DEFAULT_POLLING_INTERVAL = 100  # ms
 
@@ -15,7 +16,7 @@ def __init__(self, task_definition_name: Union[str, list]):
         self.next_task_index = 0
         self._task_definition_name_cache = None
         self._domain = None
-        self._poll_interval = DEFAULT_POLLING_INTERVAL
+        self._poll_interval = Configuration().get_poll_interval()
 
     @abc.abstractmethod
     def execute(self, task: Task) -> TaskResult:
@@ -43,7 +44,7 @@ def get_polling_interval_in_seconds(self) -> float:
         :return: float
                  Default: 100ms
         """
-        return (self.poll_interval if self.poll_interval else DEFAULT_POLLING_INTERVAL) / 1000
+        return (self.poll_interval if self.poll_interval else Configuration().get_poll_interval()) / 1000
 
     def get_task_definition_name(self) -> str:
         """