Linting

IgorChvyrov-sm · IgorChvyrov-sm · commit 12769797a9f1 · 2025-09-09T15:52:37.000+03:00
diff --git a/src/conductor/asyncio_client/automator/task_runner.py b/src/conductor/asyncio_client/automator/task_runner.py
@@ -240,16 +240,14 @@ 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(
             "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)
@@ -260,7 +258,7 @@ def __set_worker_properties(self) -> None:
                     e,
                 )
                 self.worker.poll_interval = (
-                    self.worker.get_polling_interval_in_seconds() * 1000
+                    self.worker.get_polling_interval_in_seconds()
                 )
 
     def __get_property_value_from_env(self, prop, task_type):
diff --git a/src/conductor/asyncio_client/configuration/configuration.py b/src/conductor/asyncio_client/configuration/configuration.py
@@ -121,9 +121,13 @@ def __init__(
             self.auth_secret = os.getenv("CONDUCTOR_AUTH_SECRET")
 
         # Additional worker properties with environment variable fallback
-        self.poll_interval = poll_interval or self._get_env_int("CONDUCTOR_WORKER_POLL_INTERVAL", 100)
+        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)
+        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]] = {}
diff --git a/src/conductor/asyncio_client/worker/worker.py b/src/conductor/asyncio_client/worker/worker.py
@@ -9,10 +9,12 @@
 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 WorkerInterface
diff --git a/src/conductor/asyncio_client/worker/worker_task.py b/src/conductor/asyncio_client/worker/worker_task.py
@@ -11,8 +11,8 @@
 def WorkerTask(
     task_definition_name: str,
     poll_interval: int = 100,
-    domain: str = None,
-    worker_id: str = None,
+    domain: Optional[str] = None,
+    worker_id: Optional[str] = None,
     poll_interval_seconds: int = 0,
 ):
     config = Configuration()
@@ -21,7 +21,6 @@ def WorkerTask(
     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
diff --git a/src/conductor/client/automator/task_runner.py b/src/conductor/client/automator/task_runner.py
@@ -241,16 +241,14 @@ 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(
             "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)
@@ -261,7 +259,7 @@ def __set_worker_properties(self) -> None:
                     e,
                 )
                 self.worker.poll_interval = (
-                    self.worker.get_polling_interval_in_seconds() * 1000
+                    self.worker.get_polling_interval_in_seconds()
                 )
 
     def __get_property_value_from_env(self, prop, task_type):
diff --git a/src/conductor/client/configuration/configuration.py b/src/conductor/client/configuration/configuration.py
@@ -83,9 +83,13 @@ def __init__(
         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.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)
+        self.poll_interval_seconds = poll_interval_seconds or self._get_env_float(
+            "CONDUCTOR_WORKER_POLL_INTERVAL_SECONDS", 0
+        )
 
     @property
     def debug(self):
diff --git a/src/conductor/client/worker/worker.py b/src/conductor/client/worker/worker.py
@@ -20,41 +20,41 @@
 from conductor.shared.worker.exception import NonRetryableException
 from conductor.client.worker.worker_interface import WorkerInterface
 
-ExecuteTaskFunction = Callable[
-    [
-        Union[Task, object]
-    ],
-    Union[TaskResult, object]
-]
+ExecuteTaskFunction = Callable[[Union[Task, object]], Union[TaskResult, object]]
 
-logger = logging.getLogger(
-    Configuration.get_logging_formatted_name(
-        __name__
-    )
-)
+logger = logging.getLogger(Configuration.get_logging_formatted_name(__name__))
 
 
-def is_callable_input_parameter_a_task(callable: ExecuteTaskFunction, object_type: Any) -> bool:
+def is_callable_input_parameter_a_task(
+    callable: ExecuteTaskFunction, object_type: Any
+) -> bool:
     parameters = inspect.signature(callable).parameters
     if len(parameters) != 1:
         return False
     parameter = parameters[next(iter(parameters.keys()))]
-    return parameter.annotation == object_type or parameter.annotation == parameter.empty or parameter.annotation is object  # noqa: PLR1714
+    return (
+        parameter.annotation == object_type
+        or parameter.annotation == parameter.empty
+        or parameter.annotation is object
+    )  # noqa: PLR1714
 
 
-def is_callable_return_value_of_type(callable: ExecuteTaskFunction, object_type: Any) -> bool:
+def is_callable_return_value_of_type(
+    callable: ExecuteTaskFunction, object_type: Any
+) -> bool:
     return_annotation = inspect.signature(callable).return_annotation
     return return_annotation == object_type
 
 
 class Worker(WorkerInterface):
-    def __init__(self,
-                 task_definition_name: str,
-                 execute_function: ExecuteTaskFunction,
-                 poll_interval: Optional[float] = None,
-                 domain: Optional[str] = None,
-                 worker_id: Optional[str] = None,
-                 ) -> Self:
+    def __init__(
+        self,
+        task_definition_name: str,
+        execute_function: ExecuteTaskFunction,
+        poll_interval: Optional[float] = None,
+        domain: Optional[str] = None,
+        worker_id: Optional[str] = None,
+    ) -> Self:
         super().__init__(task_definition_name)
         self.api_client = ApiClient()
         self.config = Configuration()
@@ -91,7 +91,9 @@ def execute(self, task: Task) -> TaskResult:
                         if typ in utils.simple_types:
                             task_input[input_name] = task.input_data[input_name]
                         else:
-                            task_input[input_name] = convert_from_dict_or_list(typ, task.input_data[input_name])
+                            task_input[input_name] = convert_from_dict_or_list(
+                                typ, task.input_data[input_name]
+                            )
                     elif default_value is not inspect.Parameter.empty:
                         task_input[input_name] = default_value
                     else:
@@ -118,8 +120,11 @@ def execute(self, task: Task) -> TaskResult:
                 task.task_id,
             )
 
-            task_result.logs = [TaskExecLog(
-                traceback.format_exc(), task_result.task_id, int(time.time()))]
+            task_result.logs = [
+                TaskExecLog(
+                    traceback.format_exc(), task_result.task_id, int(time.time())
+                )
+            ]
             task_result.status = TaskResultStatus.FAILED
             if len(ne.args) > 0:
                 task_result.reason_for_incompletion = ne.args[0]
@@ -130,7 +135,9 @@ def execute(self, task: Task) -> TaskResult:
             return task_result
         if not isinstance(task_result.output_data, dict):
             task_output = task_result.output_data
-            task_result.output_data = self.api_client.sanitize_for_serialization(task_output)
+            task_result.output_data = self.api_client.sanitize_for_serialization(
+                task_output
+            )
             if not isinstance(task_result.output_data, dict):
                 task_result.output_data = {"result": task_result.output_data}
 
@@ -146,11 +153,15 @@ def execute_function(self) -> ExecuteTaskFunction:
     @execute_function.setter
     def execute_function(self, execute_function: ExecuteTaskFunction) -> None:
         self._execute_function = execute_function
-        self._is_execute_function_input_parameter_a_task = is_callable_input_parameter_a_task(
-            callable=execute_function,
-            object_type=Task,
+        self._is_execute_function_input_parameter_a_task = (
+            is_callable_input_parameter_a_task(
+                callable=execute_function,
+                object_type=Task,
+            )
         )
-        self._is_execute_function_return_value_a_task_result = is_callable_return_value_of_type(
-            callable=execute_function,
-            object_type=TaskResult,
+        self._is_execute_function_return_value_a_task_result = (
+            is_callable_return_value_of_type(
+                callable=execute_function,
+                object_type=TaskResult,
+            )
         )
diff --git a/src/conductor/client/worker/worker_interface.py b/src/conductor/client/worker/worker_interface.py
@@ -44,7 +44,11 @@ def get_polling_interval_in_seconds(self) -> float:
         :return: float
                  Default: 100ms
         """
-        return (self.poll_interval if self.poll_interval else Configuration().get_poll_interval()) / 1000
+        return (
+            self.poll_interval
+            if self.poll_interval
+            else Configuration().get_poll_interval()
+        ) / 1000
 
     def get_task_definition_name(self) -> str:
         """
@@ -73,7 +77,9 @@ def clear_task_definition_name_cache(self):
     def compute_task_definition_name(self):
         if isinstance(self.task_definition_name, list):
             task_definition_name = self.task_definition_name[self.next_task_index]
-            self.next_task_index = (self.next_task_index + 1) % len(self.task_definition_name)
+            self.next_task_index = (self.next_task_index + 1) % len(
+                self.task_definition_name
+            )
             return task_definition_name
         return self.task_definition_name
 
@@ -87,7 +93,7 @@ def get_task_result_from_task(self, task: Task) -> TaskResult:
         return TaskResult(
             task_id=task.task_id,
             workflow_instance_id=task.workflow_instance_id,
-            worker_id=self.get_identity()
+            worker_id=self.get_identity(),
         )
 
     def get_domain(self) -> str:
diff --git a/src/conductor/client/worker/worker_task.py b/src/conductor/client/worker/worker_task.py
@@ -6,8 +6,13 @@
 from conductor.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,
-               poll_interval_seconds: int = 0):
+def WorkerTask(
+    task_definition_name: str,
+    poll_interval: int = 100,
+    domain: Optional[str] = None,
+    worker_id: Optional[str] = None,
+    poll_interval_seconds: int = 0,
+):
     config = Configuration()
 
     poll_interval = poll_interval or config.get_poll_interval()
@@ -20,13 +25,21 @@ def WorkerTask(task_definition_name: str, poll_interval: int = 100, domain: Opti
 
     def worker_task_func(func):
 
-        register_decorated_fn(name=task_definition_name, poll_interval=poll_interval_millis, domain=domain,
-                              worker_id=worker_id, func=func)
+        register_decorated_fn(
+            name=task_definition_name,
+            poll_interval=poll_interval_millis,
+            domain=domain,
+            worker_id=worker_id,
+            func=func,
+        )
 
         @functools.wraps(func)
         def wrapper_func(*args, **kwargs):
             if "task_ref_name" in kwargs:
-                task = SimpleTask(task_def_name=task_definition_name, task_reference_name=kwargs["task_ref_name"])
+                task = SimpleTask(
+                    task_def_name=task_definition_name,
+                    task_reference_name=kwargs["task_ref_name"],
+                )
                 kwargs.pop("task_ref_name")
                 task.input_parameters.update(kwargs)
                 return task
@@ -37,20 +50,33 @@ def wrapper_func(*args, **kwargs):
     return worker_task_func
 
 
-def worker_task(task_definition_name: str, poll_interval_millis: int = 100, domain: Optional[str] = None, worker_id: Optional[str] = None):
+def worker_task(
+    task_definition_name: str,
+    poll_interval_millis: int = 100,
+    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, poll_interval=poll_interval_millis, domain=domain,
-                              worker_id=worker_id, func=func)
+        register_decorated_fn(
+            name=task_definition_name,
+            poll_interval=poll_interval_millis,
+            domain=domain,
+            worker_id=worker_id,
+            func=func,
+        )
 
         @functools.wraps(func)
         def wrapper_func(*args, **kwargs):
             if "task_ref_name" in kwargs:
-                task = SimpleTask(task_def_name=task_definition_name, task_reference_name=kwargs["task_ref_name"])
+                task = SimpleTask(
+                    task_def_name=task_definition_name,
+                    task_reference_name=kwargs["task_ref_name"],
+                )
                 kwargs.pop("task_ref_name")
                 task.input_parameters.update(kwargs)
                 return task

Original file line number	Diff line number	Diff line change
`@@ -240,16 +240,14 @@ def __set_worker_properties(self) -> None:`
`240`	`240`	`else:`
`241`	`241`	`self.worker.domain = self.worker.get_domain()`
`242`	`242`
`243`		`- # Try both naming conventions for polling interval`
`244`	`243`	`polling_interval = self.__get_property_value_from_env(`
`245`	`244`	`"poll_interval", task_type`
`246`	`245`	`)`
`247`	`246`	`if not polling_interval:`
`248`		`- # Fallback to the "polling_interval" naming convention`
`249`	`247`	`polling_interval = self.__get_property_value_from_env(`
`250`	`248`	`"polling_interval", task_type`
`251`	`249`	`)`
`252`		`-`
	`250`	`+`
`253`	`251`	`if polling_interval:`
`254`	`252`	`try:`
`255`	`253`	`self.worker.poll_interval = float(polling_interval)`
`@@ -260,7 +258,7 @@ def __set_worker_properties(self) -> None:`
`260`	`258`	`e,`
`261`	`259`	`)`
`262`	`260`	`self.worker.poll_interval = (`
`263`		`- self.worker.get_polling_interval_in_seconds() * 1000`
	`261`	`+ self.worker.get_polling_interval_in_seconds()`
`264`	`262`	`)`
`265`	`263`
`266`	`264`	`def __get_property_value_from_env(self, prop, task_type):`
Original file line number	Diff line number	Diff line change
`@@ -241,16 +241,14 @@ def __set_worker_properties(self) -> None:`
`241`	`241`	`else:`
`242`	`242`	`self.worker.domain = self.worker.get_domain()`
`243`	`243`
`244`		`- # Try both naming conventions for polling interval`
`245`	`244`	`polling_interval = self.__get_property_value_from_env(`
`246`	`245`	`"poll_interval", task_type`
`247`	`246`	`)`
`248`	`247`	`if not polling_interval:`
`249`		`- # Fallback to the "polling_interval" naming convention`
`250`	`248`	`polling_interval = self.__get_property_value_from_env(`
`251`	`249`	`"polling_interval", task_type`
`252`	`250`	`)`
`253`		`-`
	`251`	`+`
`254`	`252`	`if polling_interval:`
`255`	`253`	`try:`
`256`	`254`	`self.worker.poll_interval = float(polling_interval)`
`@@ -261,7 +259,7 @@ def __set_worker_properties(self) -> None:`
`261`	`259`	`e,`
`262`	`260`	`)`
`263`	`261`	`self.worker.poll_interval = (`
`264`		`- self.worker.get_polling_interval_in_seconds() * 1000`
	`262`	`+ self.worker.get_polling_interval_in_seconds()`
`265`	`263`	`)`
`266`	`264`
`267`	`265`	`def __get_property_value_from_env(self, prop, task_type):`