Lint fixes for B, PERF, RUF and C4 errors (#311)

Author: IgorChvyrov-sm

src/conductor/client/ai/integrations.py

@@ -5,9 +5,6 @@
 from typing import Optional
 
 class IntegrationConfig(ABC):
-    def __init__(self):
-        pass
-
     @abstractmethod
     def to_dict(self) -> dict:
         pass

src/conductor/client/ai/orchestrator.py

@@ -47,10 +47,10 @@ 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: Optional[List[str]] = [], max_tokens: Optional[int] = 100,
+                             stop_words: List[str] = None, max_tokens: int = 100,
                              temperature: int = 0,
                              top_p: int = 1):
-
+        stop_words = stop_words or []
         return self.prompt_client.test_prompt(text, variables, ai_integration, text_complete_model, temperature, top_p,
                                               stop_words)
 

src/conductor/client/automator/task_handler.py

@@ -46,12 +46,13 @@ def register_decorated_fn(name: str, poll_interval: int, domain: str, worker_id:
 class TaskHandler:
     def __init__(
             self,
-            workers: List[WorkerInterface] = [],
+            workers: Optional[List[WorkerInterface]] = None,
             configuration: Optional[Configuration] = None,
             metrics_settings: Optional[MetricsSettings] = None,
             scan_for_annotated_workers: bool = True,
             import_modules: Optional[List[str]] = None
     ):
+        workers = workers or []
         self.logger_process, self.queue = _setup_logging_queue(configuration)
 
         # imports
@@ -62,8 +63,6 @@ def __init__(
                 logger.info(f'loading module {module}')
                 importlib.import_module(module)
 
-        if workers is None:
-            workers = []
         elif not isinstance(workers, list):
             workers = [workers]
         if scan_for_annotated_workers is True:

src/conductor/client/automator/task_runner.py

@@ -58,18 +58,18 @@ def run(self) -> None:
                     f'interval {self.worker.get_polling_interval_in_seconds()}')
 
         while True:
-            try:
-                self.run_once()
-            except Exception:
-                pass
+            self.run_once()
 
     def run_once(self) -> None:
-        task = self.__poll_task()
-        if task is not None and task.task_id is not None:
-            task_result = self.__execute_task(task)
-            self.__update_task(task_result)
-        self.__wait_for_polling_interval()
-        self.worker.clear_task_definition_name_cache()
+        try:
+            task = self.__poll_task()
+            if task is not None and task.task_id is not None:
+                task_result = self.__execute_task(task)
+                self.__update_task(task_result)
+            self.__wait_for_polling_interval()
+            self.worker.clear_task_definition_name_cache()
+        except Exception as e:
+            pass
 
     def __poll_task(self) -> Task:
         task_definition_name = self.worker.get_task_definition_name()

src/conductor/client/automator/utils.py

@@ -72,12 +72,10 @@ def convert_from_dict(cls: type, data: dict) -> object:
             else:
                 kwargs[member] = members[member].default
         elif str(typ).startswith('typing.List[') or str(typ).startswith('typing.Set[') or str(typ).startswith('list['):
-            values = []
             generic_type = object
             if len(generic_types) > 0:
                 generic_type = generic_types[0]
-            for val in data[member]:
-                values.append(get_value(generic_type, val))
+            values = [get_value(generic_type, item) for item in data[member]]
             kwargs[member] = values
         elif (str(typ).startswith('dict[') or
               str(typ).startswith('typing.Dict[') or
@@ -112,8 +110,8 @@ def get_value(typ: type, val: object) -> object:
         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)
+        for item in val:
+            converted = get_value(type(item), item)
             values.append(converted)
         return values
     elif str(typ).startswith('dict[') or str(typ).startswith(

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

@@ -14,7 +14,7 @@
 
 
 def get_default_temporary_folder() -> str:
-    return f'{str(Path.home())}/tmp/'
+    return f'{Path.home()!s}/tmp/'
 
 
 class MetricsSettings:

src/conductor/client/event/queue/queue_configuration.py

@@ -1,12 +1,12 @@
-from abc import ABC
-from typing import Any, Dict
+from abc import ABC, abstractmethod
+from typing import Any, ClassVar, Dict
 
 from conductor.client.event.queue.queue_worker_configuration import QueueWorkerConfiguration
 
 
 class QueueConfiguration(ABC):
-    WORKER_CONSUMER_KEY = "consumer"
-    WORKER_PRODUCER_KEY = "producer"
+    WORKER_CONSUMER_KEY: ClassVar[str] = "consumer"
+    WORKER_PRODUCER_KEY: ClassVar[str] = "producer"
 
     def __init__(self, queue_name: str, queue_type: str):
         self.queue_name = queue_name
@@ -19,5 +19,6 @@ def add_consumer(self, worker_configuration: QueueWorkerConfiguration) -> None:
     def add_producer(self, worker_configuration: QueueWorkerConfiguration) -> None:
         self.worker_configuration[self.WORKER_PRODUCER_KEY] = worker_configuration
 
+    @abstractmethod
     def get_worker_configuration(self) -> Dict[str, Any]:
         raise NotImplementedError

src/conductor/client/exceptions/api_exception_handler.py

@@ -46,11 +46,11 @@ def inner_function(*args, **kwargs):
     return inner_function
 
 
-def for_all_methods(decorator, exclude=[]):
+def for_all_methods(decorator, exclude=None):
     def decorate(cls):
+        exclude_local = [] if exclude is None else exclude
         for attr in cls.__dict__:
-            if callable(getattr(cls, attr)) and attr not in exclude:
+            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

@@ -1,6 +1,7 @@
 import datetime
 import logging
 import re
+from typing import Any, ClassVar, Dict, Tuple
 
 import six
 from requests.structures import CaseInsensitiveDict
@@ -17,10 +18,10 @@
 
 
 class ObjectMapper(object):
-    PRIMITIVE_TYPES = (float, bool, bytes, six.text_type) + six.integer_types
-    NATIVE_TYPES_MAPPING = {
+    PRIMITIVE_TYPES: ClassVar[Tuple[Any, ...]] = (float, bool, bytes, six.text_type, *six.integer_types)
+    NATIVE_TYPES_MAPPING: ClassVar[Dict[str, Any]] = {
         'int': int,
-        'long': int if six.PY3 else long,  # noqa: F821
+        'long': int if six.PY3 else long,  # noqa: F821, RUF100
         'float': float,
         'str': str,
         'bool': bool,
@@ -137,11 +138,11 @@ def __deserialize_date(self, string):
             return parse(string).date()
         except ImportError:
             return string
-        except ValueError:
+        except ValueError as err:
             raise rest.ApiException(
                 status=0,
                 reason="Failed to parse `{0}` as date object".format(string)
-            )
+            ) from err
 
     def __deserialize_datatime(self, string):
         """Deserializes string to datetime.
@@ -156,14 +157,14 @@ def __deserialize_datatime(self, string):
             return parse(string)
         except ImportError:
             return string
-        except ValueError:
+        except ValueError as err:
             raise rest.ApiException(
                 status=0,
                 reason=(
                     "Failed to parse `{0}` as datetime object"
                     .format(string)
                 )
-            )
+            ) from err
 
     def __hasattr(self, object, name):
         return name in object.__class__.__dict__

src/conductor/client/integration_client.py

@@ -42,58 +42,75 @@ def delete_integration_api(self, api_name:str, integration_name:str):
         """Delete a specific integration api for a given integration"""
         pass
 
+    @abstractmethod
     def delete_integration(self, integration_name:str):
         """Delete an integration"""
         pass
 
+    @abstractmethod
     def get_integration_api(self, api_name:str, integration_name:str) -> IntegrationApi:
         pass
 
+    @abstractmethod
     def get_integration_apis(self, integration_name:str) -> List[IntegrationApi]:
         pass
 
+    @abstractmethod
     def get_integration(self, integration_name:str) -> Integration:
         pass
 
+    @abstractmethod
     def get_integrations(self) -> List[Integration]:
         """Returns the list of all the available integrations"""
         pass
 
+    @abstractmethod
     def get_prompts_with_integration(self, ai_integration:str, model_name:str) -> List[PromptTemplate]:
         pass
 
+    @abstractmethod
     def get_token_usage_for_integration(self, name, integration_name) -> int:
         pass
 
+    @abstractmethod
     def get_token_usage_for_integration_provider(self, name) -> dict:
         pass
 
+    @abstractmethod
     def register_token_usage(self, body, name, integration_name):
         pass
 
+    @abstractmethod
     def save_integration_api(self, integration_name, api_name, api_details: IntegrationApiUpdate):
         pass
 
+    @abstractmethod
     def save_integration(self, integration_name, integration_details: IntegrationUpdate):
         pass
 
     # Tags
 
+    @abstractmethod
     def delete_tag_for_integration(self, body, tag_name, integration_name):
         """Delete an integration"""
         pass
 
+    @abstractmethod
     def delete_tag_for_integration_provider(self, body, name):
         pass
 
+    @abstractmethod
     def put_tag_for_integration(self, body, name, integration_name):
         pass
 
+    @abstractmethod
     def put_tag_for_integration_provider(self, body, name):
         pass
 
+    @abstractmethod
     def get_tags_for_integration(self, name, integration_name):
         pass
 
+    @abstractmethod
     def get_tags_for_integration_provider(self, name):
         pass