Refactor final linter fixes (#313)

Author: IgorChvyrov-sm

.pre-commit-config.yaml

@@ -3,4 +3,4 @@ repos:
     rev: v0.12.3
     hooks:
       - id: ruff
-        args: ["--exit-zero"]
+        args: ["."]

pyproject.toml

@@ -87,8 +87,6 @@ select = [
   "TID",
   # implicit string concatenation
   "ISC",
-  # flake8-type-checking
-  "TC",
   # comprehensions
   "C4",
   # pygrep-hooks
@@ -128,7 +126,9 @@ ignore = [
     # too-many-branches
     "PLR0912",
     # too-many-return-statements
-    "PLR0911"
+    "PLR0911",
+    # flake8-type-checking (150 errors)
+    "TC"
 ]
 
 [tool.ruff.lint.isort]

src/conductor/client/ai/orchestrator.py

@@ -50,7 +50,7 @@ def associate_prompt_template(self, name: str, ai_integration: str, ai_models: L
     def test_prompt_template(self, text: str, variables: dict,
                              ai_integration: str,
                              text_complete_model: str,
-                             stop_words: List[str] = None, max_tokens: int = 100,
+                             stop_words: Optional[List[str]] = None, max_tokens: int = 100,
                              temperature: int = 0,
                              top_p: int = 1):
         stop_words = stop_words or []

src/conductor/client/automator/task_handler.py

@@ -29,12 +29,12 @@
             set_start_method("fork")
         _mp_fork_set = True
     except Exception as e:
-        logger.info(f"error when setting multiprocessing.set_start_method - maybe the context is set {e.args}")
+        logger.info("error when setting multiprocessing.set_start_method - maybe the context is set %s", e.args)
     if platform == "darwin":
         os.environ["no_proxy"] = "*"
 
 def register_decorated_fn(name: str, poll_interval: int, domain: str, worker_id: str, func):
-    logger.info(f"decorated {name}")
+    logger.info("decorated %s", name)
     _decorated_functions[(name, domain)] = {
         "func": func,
         "poll_interval": poll_interval,
@@ -60,24 +60,24 @@ def __init__(
         importlib.import_module("conductor.client.worker.worker_task")
         if import_modules is not None:
             for module in import_modules:
-                logger.info(f"loading module {module}")
+                logger.info("loading module %s", module)
                 importlib.import_module(module)
 
         elif not isinstance(workers, list):
             workers = [workers]
         if scan_for_annotated_workers is True:
             for (task_def_name, domain), record in _decorated_functions.items():
-                fn = record['func']
-                worker_id = record['worker_id']
-                poll_interval = record['poll_interval']
+                fn = record["func"]
+                worker_id = record["worker_id"]
+                poll_interval = record["poll_interval"]
 
                 worker = Worker(
                     task_definition_name=task_def_name,
                     execute_function=fn,
                     worker_id=worker_id,
                     domain=domain,
                     poll_interval=poll_interval)
-                logger.info(f"created worker with name={task_def_name} and domain={domain}")
+                logger.info("created worker with name=%s and domain=%s", task_def_name, domain)
                 workers.append(worker)
 
         self.__create_task_runner_processes(workers, configuration, metrics_settings)
@@ -156,7 +156,7 @@ def __start_task_runner_processes(self):
         for task_runner_process in self.task_runner_processes:
             task_runner_process.start()
             n = n + 1
-        logger.info(f"Started {n} TaskRunner process")
+        logger.info("Started %s TaskRunner process", n)
 
     def __join_metrics_provider_process(self):
         if self.metrics_provider_process is None:
@@ -180,12 +180,12 @@ def __stop_process(self, process: Process):
         if process is None:
             return
         try:
-            logger.debug(f"Terminating process: {process.pid}")
+            logger.debug("Terminating process: %s", process.pid)
             process.terminate()
         except Exception as e:
-            logger.debug(f"Failed to terminate process: {process.pid}, reason: {e}")
+            logger.debug("Failed to terminate process: %s, reason: %s", process.pid, e)
             process.kill()
-            logger.debug(f"Killed process: {process.pid}")
+            logger.debug("Killed process: %s", process.pid)
 
 
 # Setup centralized logging queue

src/conductor/client/automator/task_runner.py

@@ -54,8 +54,12 @@ def run(self) -> None:
             logger.setLevel(logging.DEBUG)
 
         task_names = ",".join(self.worker.task_definition_names)
-        logger.info(f"Polling task {task_names} with domain {self.worker.get_domain()} with polling "
-                    f"interval {self.worker.get_polling_interval_in_seconds()}")
+        logger.info(
+            "Polling task %s with domain %s with polling interval %s",
+            task_names,
+            self.worker.get_domain(),
+            self.worker.get_polling_interval_in_seconds()
+        )
 
         while True:
             self.run_once()
@@ -68,13 +72,13 @@ def run_once(self) -> None:
                 self.__update_task(task_result)
             self.__wait_for_polling_interval()
             self.worker.clear_task_definition_name_cache()
-        except Exception as e:
+        except Exception:
             pass
 
     def __poll_task(self) -> Task:
         task_definition_name = self.worker.get_task_definition_name()
         if self.worker.paused():
-            logger.debug(f"Stop polling task for: {task_definition_name}")
+            logger.debug("Stop polling task for: %s", task_definition_name)
             return None
         if self.metrics_collector is not None:
             self.metrics_collector.increment_task_poll(
@@ -104,24 +108,29 @@ def __poll_task(self) -> Task:
             if self.metrics_collector is not None:
                 self.metrics_collector.increment_task_poll_error(task_definition_name, type(e))
             logger.error(
-                f"Failed to poll task for: {task_definition_name}, reason: {traceback.format_exc()}"
+                "Failed to poll task for: %s, reason: %s",
+                task_definition_name,
+                traceback.format_exc()
             )
             return None
         if task is not None:
             logger.debug(
-                f"Polled task: {task_definition_name}, worker_id: {self.worker.get_identity()}, domain: {self.worker.get_domain()}")
+                "Polled task: %s, worker_id: %s, domain: %s",
+                task_definition_name,
+                self.worker.get_identity(),
+                self.worker.get_domain()
+            )
         return task
 
     def __execute_task(self, task: Task) -> TaskResult:
         if not isinstance(task, Task):
             return None
         task_definition_name = self.worker.get_task_definition_name()
         logger.debug(
-            "Executing task, id: {task_id}, workflow_instance_id: {workflow_instance_id}, task_definition_name: {task_definition_name}".format(
-                task_id=task.task_id,
-                workflow_instance_id=task.workflow_instance_id,
-                task_definition_name=task_definition_name
-            )
+            "Executing task, id: %s, workflow_instance_id: %s, task_definition_name: %s",
+            task.task_id,
+            task.workflow_instance_id,
+            task_definition_name
         )
         try:
             start_time = time.time()
@@ -138,11 +147,10 @@ def __execute_task(self, task: Task) -> TaskResult:
                     sys.getsizeof(task_result)
                 )
             logger.debug(
-                "Executed task, id: {task_id}, workflow_instance_id: {workflow_instance_id}, task_definition_name: {task_definition_name}".format(
-                    task_id=task.task_id,
-                    workflow_instance_id=task.workflow_instance_id,
-                    task_definition_name=task_definition_name
-                )
+                "Executed task, id: %s, workflow_instance_id: %s, task_definition_name: %s",
+                task.task_id,
+                task.workflow_instance_id,
+                task_definition_name
             )
         except Exception as e:
             if self.metrics_collector is not None:
@@ -159,12 +167,12 @@ def __execute_task(self, task: Task) -> TaskResult:
             task_result.logs = [TaskExecLog(
                 traceback.format_exc(), task_result.task_id, int(time.time()))]
             logger.error(
-                "Failed to execute task, id: {task_id}, workflow_instance_id: {workflow_instance_id}, task_definition_name: {task_definition_name}, reason: {reason}".format(
-                    task_id=task.task_id,
-                    workflow_instance_id=task.workflow_instance_id,
-                    task_definition_name=task_definition_name,
-                    reason=traceback.format_exc()
-                )
+                "Failed to execute task, id: %s, workflow_instance_id: %s, "
+                "task_definition_name: %s, reason: %s",
+                task.task_id,
+                task.workflow_instance_id,
+                task_definition_name,
+                traceback.format_exc()
             )
         return task_result
 
@@ -173,11 +181,10 @@ def __update_task(self, task_result: TaskResult):
             return None
         task_definition_name = self.worker.get_task_definition_name()
         logger.debug(
-            "Updating task, id: {task_id}, workflow_instance_id: {workflow_instance_id}, task_definition_name: {task_definition_name}".format(
-                task_id=task_result.task_id,
-                workflow_instance_id=task_result.workflow_instance_id,
-                task_definition_name=task_definition_name
-            )
+            "Updating task, id: %s, workflow_instance_id: %s, task_definition_name: %s",
+            task_result.task_id,
+            task_result.workflow_instance_id,
+            task_definition_name
         )
         for attempt in range(4):
             if attempt > 0:
@@ -186,12 +193,11 @@ def __update_task(self, task_result: TaskResult):
             try:
                 response = self.task_client.update_task(body=task_result)
                 logger.debug(
-                    "Updated task, id: {task_id}, workflow_instance_id: {workflow_instance_id}, task_definition_name: {task_definition_name}, response: {response}".format(
-                        task_id=task_result.task_id,
-                        workflow_instance_id=task_result.workflow_instance_id,
-                        task_definition_name=task_definition_name,
-                        response=response
-                    )
+                    "Updated task, id: %s, workflow_instance_id: %s, task_definition_name: %s, response: %s",
+                    task_result.task_id,
+                    task_result.workflow_instance_id,
+                    task_definition_name,
+                    response
                 )
                 return response
             except Exception as e:
@@ -200,13 +206,11 @@ def __update_task(self, task_result: TaskResult):
                         task_definition_name, type(e)
                     )
                 logger.error(
-                    "Failed to update task, id: {task_id}, workflow_instance_id: {workflow_instance_id}, "
-                    "task_definition_name: {task_definition_name}, reason: {reason}".format(
-                        task_id=task_result.task_id,
-                        workflow_instance_id=task_result.workflow_instance_id,
-                        task_definition_name=task_definition_name,
-                        reason=traceback.format_exc()
-                    )
+                    "Failed to update task, id: %s, workflow_instance_id: %s, task_definition_name: %s, reason: %s",
+                    task_result.task_id,
+                    task_result.workflow_instance_id,
+                    task_definition_name,
+                    traceback.format_exc()
                 )
         return None
 
@@ -230,14 +234,14 @@ def __set_worker_properties(self) -> None:
             try:
                 self.worker.poll_interval = float(polling_interval)
             except Exception:
-                logger.error(f"error reading and parsing the polling interval value {polling_interval}")
+                logger.error("error reading and parsing the polling interval value %s", polling_interval)
                 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: {0}.".format(str(e)))
+                logger.error("Exception in reading polling interval from environment variable: %s", e)
 
     def __get_property_value_from_env(self, prop, task_type):
         """

src/conductor/client/automator/utils.py

@@ -54,7 +54,7 @@ def convert_from_dict(cls: type, data: dict) -> object:
     if not ((str(typ).startswith("dict[") or
              str(typ).startswith("typing.Dict[") or
              str(typ).startswith("requests.structures.CaseInsensitiveDict[") or
-             typ == dict or str(typ).startswith("OrderedDict["))):
+             typ is dict or str(typ).startswith("OrderedDict["))):
         data = {}
 
     members = inspect.signature(cls.__init__).parameters
@@ -82,7 +82,7 @@ def convert_from_dict(cls: type, data: dict) -> object:
         elif (str(typ).startswith("dict[") or
               str(typ).startswith("typing.Dict[") or
               str(typ).startswith("requests.structures.CaseInsensitiveDict[") or
-              typ == dict or str(typ).startswith("OrderedDict[")):
+              typ is dict or str(typ).startswith("OrderedDict[")):
 
             values = {}
             generic_type = object
@@ -111,13 +111,10 @@ def get_value(typ: type, val: object) -> object:
     if typ in simple_types:
         return val
     elif str(typ).startswith("typing.List[") or str(typ).startswith("typing.Set[") or str(typ).startswith("list["):
-        values = []
-        for val in val:
-            converted = get_value(type(val), val)
-            values.append(converted)
+        values = [get_value(type(item), item) for item in val]
         return values
     elif str(typ).startswith("dict[") or str(typ).startswith(
-            "typing.Dict[") or str(typ).startswith("requests.structures.CaseInsensitiveDict[") or typ == dict:
+            "typing.Dict[") or str(typ).startswith("requests.structures.CaseInsensitiveDict[") or typ is dict:
         values = {}
         for k in val:
             v = val[k]

src/conductor/client/configuration/settings/metrics_settings.py

@@ -1,9 +1,10 @@
 from __future__ import annotations
 import logging
 import os
-from typing import Optional
 from pathlib import Path
 
+from typing import Optional
+
 from conductor.client.configuration.configuration import Configuration
 
 logger = logging.getLogger(
@@ -20,7 +21,7 @@ def get_default_temporary_folder() -> str:
 class MetricsSettings:
     def __init__(
             self,
-            directory: str = None,
+            directory: Optional[str] = None,
             file_name: str = "metrics.log",
             update_interval: float = 0.1):
         if directory is None:
@@ -35,6 +36,6 @@ def __set_dir(self, dir: str) -> None:
                 os.mkdir(dir)
             except Exception as e:
                 logger.warning(
-                    "Failed to create metrics temporary folder, reason: ", e)
+                    "Failed to create metrics temporary folder, reason: %s", e)
 
         self.directory = dir

src/conductor/client/exceptions/api_exception_handler.py

@@ -59,4 +59,4 @@ def decorate(cls):
             if callable(getattr(cls, attr)) and attr not in exclude_local:
                 setattr(cls, attr, decorator(getattr(cls, attr)))
         return cls
-    return decorate
+    return decorate

src/conductor/client/helpers/helper.py

@@ -2,7 +2,7 @@
 import logging
 import re
 from dateutil.parser import parse
-from typing import Any, ClassVar, Dict, Tuple
+from typing import ClassVar, Dict, Tuple
 
 import six
 from requests.structures import CaseInsensitiveDict
@@ -19,10 +19,10 @@
 
 
 class ObjectMapper(object):
-    PRIMITIVE_TYPES = (float, bool, bytes, six.text_type) + six.integer_types
-    NATIVE_TYPES_MAPPING = {
+    PRIMITIVE_TYPES: ClassVar[Tuple] = (float, bool, bytes, six.text_type, *six.integer_types)
+    NATIVE_TYPES_MAPPING: ClassVar[Dict] = {
         "int": int,
-        "long": int if six.PY3 else long,  # noqa: F821
+        "long": int if six.PY3 else long,  # noqa: F821, YTT202
         "float": float,
         "str": str,
         "bool": bool,
@@ -68,7 +68,7 @@ def __deserialize(self, data, klass):
         if data is None:
             return None
 
-        if type(klass) == str:
+        if isinstance(klass, str):
             if klass.startswith("list["):
                 sub_kls = re.match(r"list\[(.*)\]", klass).group(1)
 

src/conductor/client/orkes/orkes_workflow_client.py

@@ -202,7 +202,7 @@ def get_by_correlation_ids(
     def remove_workflow(self, workflow_id: str):
         self.workflowResourceApi.delete(workflow_id)
 
-    def update_variables(self, workflow_id: str, variables: Dict[str, object] = None) -> None:
+    def update_variables(self, workflow_id: str, variables: Optional[Dict[str, object]] = None) -> None:
         variables = variables or {}
         self.workflowResourceApi.update_workflow_state(variables, workflow_id)