Merge branch 'main' into refactor_lint_fixes_PLR_PLW_PLC_PYI_TC

Author: IgorChvyrov-sm

src/conductor/__init__.py

@@ -1 +1 @@
-__version__ = "1.1.10"
+__version__ = "1.1.10"

src/conductor/client/ai/integrations.py

@@ -2,12 +2,9 @@
 
 import os
 from abc import ABC, abstractmethod
-
+from typing import Optional
 
 class IntegrationConfig(ABC):
-    def __init__(self):
-        pass
-
     @abstractmethod
     def to_dict(self) -> dict:
         pass
@@ -29,7 +26,7 @@ def to_dict(self) -> dict:
 
 class OpenAIConfig(IntegrationConfig):
 
-    def __init__(self, api_key: str = None) -> None:
+    def __init__(self, api_key: Optional[str] = None) -> None:
         if api_key is None:
             api_key = os.getenv('OPENAI_API_KEY')
         self.api_key = api_key
@@ -55,7 +52,7 @@ def to_dict(self) -> dict:
 
 class PineconeConfig(IntegrationConfig):
 
-    def __init__(self, api_key: str = None, endpoint: str = None, environment: str = None, project_name: str = None) -> None:
+    def __init__(self, api_key: Optional[str] = None, endpoint: Optional[str] = None, environment: Optional[str] = None, project_name: Optional[str] = None) -> None:
         if api_key is None:
             self.api_key = os.getenv('PINECONE_API_KEY')
         else:

src/conductor/client/ai/orchestrator.py

@@ -50,10 +50,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)
 
@@ -76,8 +76,8 @@ def add_ai_integration(self, ai_integration_name: str, provider: LLMProvider, mo
             if existing_integration_api is None or overwrite:
                 self.integration_client.save_integration_api(ai_integration_name, model, api_details)
 
-    def add_vector_store(self, db_integration_name: str, provider: VectorDB, indices: List[str],config: IntegrationConfig,
-                         description: str = None,overwrite : bool = False):
+    def add_vector_store(self, db_integration_name: str, provider: VectorDB, indices: List[str], config: IntegrationConfig,
+                         description: Optional[str] = None, overwrite: bool = False):
         vector_db = IntegrationUpdate()
         vector_db.configuration = config.to_dict()
         vector_db.type = provider.value

src/conductor/client/authorization_client.py

@@ -1,3 +1,4 @@
+from __future__ import annotations
 from abc import ABC, abstractmethod
 from typing import Dict, List, Optional
 from conductor.client.orkes.models.metadata_tag import MetadataTag

src/conductor/client/automator/task_handler.py

@@ -1,9 +1,10 @@
+from __future__ import annotations
 import importlib
 import logging
 import os
-from multiprocessing import Process, freeze_support, Queue, set_start_method, get_context
+from multiprocessing import Process, freeze_support, Queue, set_start_method
 from sys import platform
-from typing import List
+from typing import List, Optional
 
 from conductor.client.automator.task_runner import TaskRunner
 from conductor.client.configuration.configuration import Configuration
@@ -45,12 +46,13 @@ def register_decorated_fn(name: str, poll_interval: int, domain: str, worker_id:
 class TaskHandler:
     def __init__(
             self,
-            workers: List[WorkerInterface] = [],
-            configuration: Configuration = None,
-            metrics_settings: MetricsSettings = None,
+            workers: Optional[List[WorkerInterface]] = None,
+            configuration: Optional[Configuration] = None,
+            metrics_settings: Optional[MetricsSettings] = None,
             scan_for_annotated_workers: bool = True,
-            import_modules: List[str] = None
+            import_modules: Optional[List[str]] = None
     ):
+        workers = workers or []
         self.logger_process, self.queue = _setup_logging_queue(configuration)
 
         # imports
@@ -61,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 as e:
-                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()
@@ -229,14 +229,13 @@ def __set_worker_properties(self) -> None:
         if polling_interval:
             try:
                 self.worker.poll_interval = float(polling_interval)
-            except Exception as e:
+            except Exception:
                 logger.error(f'error reading and parsing the polling interval value {polling_interval}')
                 self.worker.poll_interval = self.worker.get_polling_interval_in_seconds()
 
         if polling_interval:
             try:
                 self.worker.poll_interval = float(polling_interval)
-                polling_interval_initialized = True
             except Exception as e:
                 logger.error("Exception in reading polling interval from environment variable: {0}.".format(str(e)))
 

src/conductor/client/automator/utils.py

@@ -1,3 +1,4 @@
+from __future__ import annotations
 import dataclasses
 import datetime
 import inspect
@@ -43,7 +44,7 @@ def convert_from_dict(cls: type, data: dict) -> object:
     if data is None:
         return data
 
-    if type(data) == cls:
+    if isinstance(data, cls):
         return data
 
     if dataclasses.is_dataclass(cls):
@@ -53,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
@@ -71,17 +72,15 @@ 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
               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
@@ -91,7 +90,7 @@ def convert_from_dict(cls: type, data: dict) -> object:
                 v = data[member][k]
                 values[k] = get_value(generic_type, v)
             kwargs[member] = values
-        elif typ == inspect.Parameter.empty:
+        elif typ is inspect.Parameter.empty:
             if inspect.Parameter.VAR_KEYWORD == members[member].kind:
                 if type(data) in dict_types:
                     kwargs.update(data)
@@ -113,7 +112,7 @@ def get_value(typ: type, val: object) -> object:
         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/configuration.py

@@ -1,6 +1,8 @@
+from __future__ import annotations
 import logging
 import os
 import time
+from typing import Optional
 
 from conductor.client.configuration.settings.authentication_settings import AuthenticationSettings
 
@@ -10,10 +12,10 @@ class Configuration:
 
     def __init__(
             self,
-            base_url: str = None,
+            base_url: Optional[str] = None,
             debug: bool = False,
             authentication_settings: AuthenticationSettings = None,
-            server_api_url: str = None,
+            server_api_url: Optional[str] = None,
             auth_token_ttl_min: int = 45
     ):
         if server_api_url is not None:
@@ -138,7 +140,7 @@ def ui_host(self):
         """
         return self.__ui_host
 
-    def apply_logging_config(self, log_format : str = None, level = None):
+    def apply_logging_config(self, log_format : Optional[str] = None, level = None):
         if log_format is None:
             log_format = self.logger_format
         if level is None:

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

@@ -1,5 +1,7 @@
+from __future__ import annotations
 import logging
 import os
+from typing import Optional
 from pathlib import Path
 
 from conductor.client.configuration.configuration import Configuration
@@ -12,13 +14,13 @@
 
 
 def get_default_temporary_folder() -> str:
-    return f'{str(Path.home())}/tmp/'
+    return f'{Path.home()!s}/tmp/'
 
 
 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:

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