diff --git a/smartassist/src/utils.py b/smartassist/src/utils.py index bc0e1df..b7aa187 100644 --- a/smartassist/src/utils.py +++ b/smartassist/src/utils.py @@ -4,6 +4,8 @@ import logging import json import requests +from typing import Optional +from enums import LogLevel class GlobalState: """ @@ -15,7 +17,12 @@ class GlobalState: """ _instance = None # Private class attribute to hold the single instance of the class - def __new__(cls): + def __new__(cls) -> 'GlobalState': + """ + Create a new instance of the GlobalState class. + + This is a singleton implementation, so only one instance will be created. + """ if cls._instance is None: cls._instance = super(GlobalState, cls).__new__(cls) cls._instance.log_level = 'INFO' # Default logging level @@ -36,8 +43,27 @@ class GlobalState: return cls._instance - def configure_logging(self, level=None): - """Set up logging for the project.""" + # def configure_logging(self, level: Optional[LogLevel] = None) -> None: + # """ + # Configure the logging system for this project. + + # Args: + # level (LogLevel): The log level to use. If None, uses the default log level set in `self.log_level`. + + # Notes: + # This method sets up logging for the project and logs a message at the debug level indicating the effective log level. + # """ + def configure_logging(self, level: Optional[str] = None) -> None: + + """ + Configure the logging system for this project. + + Args: + level (str): The log level to use. Can be one of the standard Python log levels (e.g., 'DEBUG', 'INFO', 'WARNING', etc.). If None, uses the default log level set in `self.log_level`. + + Notes: + This method sets up logging for the project and logs a message at the debug level indicating the effective log level. + """ if level is None: level = self.log_level # numeric_level = getattr(logging, level.upper()) # Convert string to numeric level @@ -45,106 +71,220 @@ class GlobalState: self.logger.setLevel(numeric_level) self.logger.debug(f"utils.py -- configure_logging(): effective log level is {level} which is {self.logger.getEffectiveLevel()}") - def set_log_level(self, level = 'INFO'): - """Set the logging level.""" + # def set_log_level(self, level: LogLevel) -> None: + # """ + # Set the log level for this project. + + # Args: + # level (LogLevel): The new log level to use. Can be one of the evels defined in enum.py (e.g., DEBUG, INFO, WARNING, CRITICAL etc.). + + # Notes: + # This method updates the `self.log_level` attribute and calls `configure_logging()` to apply the change. + # """ + def set_log_level(self, level: str = 'INFO') -> None: + """ + Set the log level for this project. + + Args: + level (str): The new log level to use. Can be one of the standard Python log levels (e.g., 'DEBUG', 'INFO', 'WARNING', etc.). + + Notes: + This method updates the `self.log_level` attribute and calls `configure_logging()` to apply the change. + """ self.log_level = level self.configure_logging() - def get_log_level(self): - """Getter for log_level attribute.""" + def get_log_level(self) -> str: + """ + Get the current log level. + + Returns: + str: The current log level (e.g., 'DEBUG', 'INFO', 'WARNING', etc.). + """ return self.log_level - def get_effective_log_level(self): - """Getter for effective log level of loggerattribute.""" + def get_effective_log_level(self) -> int: + """ + Get the effective log level of the logger. + + Returns: + int: The numeric value of the effective log level. + """ return self.logger.getEffectiveLevel() - def getLogger(self, module_name = None): - """Return a logger based on the module name.""" + def get_logger(self, module_name: Optional[str] = None) -> logging.Logger: + + """ + Get a logger instance based on the module name. + + Args: + module_name (str): The name of the module to get a logger for. If None, uses the current module name (`__name__`). + + Returns: + Logger: A logger instance configured for the specified module. + """ if module_name is None: module_name = __name__ logger = logging.getLogger(module_name) return logger - def set_host_url(self, url="http://localhost:11434"): - """Set the host url to which LLM requests are sent""" - self.host = url + def set_host_url(self, url: str = "http://localhost:11434") -> None: + """ + Set the URL of the host to which LLM requests are sent. - def get_host_url(self): - """Get the url for the currently used host for LLMs""" + Args: + url (str): The new URL to use. Defaults to 'http://localhost:11434' if not specified. + """ + self.host_url = url + + def get_host_url(self) -> str: + """ + Get the current URL of the host used for LLMs. + + Returns: + str: The current URL of the host. + """ return self.host_url - def set_llm(self, model_name="phi3:mini"): - """Set LLM for queries""" + def set_llm(self, model_name: str = "phi3:mini") -> None: + """ + Set the LLM to use for queries. + + Args: + model_name (str): The name of the LLM to use. Defaults to 'phi3:mini' if not specified. + """ self.llm = model_name - def get_llm(self): - """Getter for which LLM is used for queries""" + def get_llm(self) -> str: + """ + Get the current LLM used for queries. + + Returns: + str: The name of the current LLM. + """ return self.llm - def set_backend(self, backend=None): - """Set backend server that web clients connect to""" - self.backend = backend + def set_backend(self, backend: Optional[dict] = None) -> None: - def get_backend(self): - """Getter for backend server that web clients connect to""" + """ + Set the backend server that web clients connect to. + + Args: + backend (dict): A dictionary containing information about the backend server. If None, resets the backend server to its default value. + """ + self.backend = backend + + def get_backend(self) -> dict: + """ + Get the current backend server used by web clients. + + Returns: + dict: A dictionary containing information about the current backend server. + """ return self.backend - def get_backend_api_ep(self): - """Getter for backend API endpoint""" + def get_backend_api_ep(self) -> str: + """ + Get the API endpoint of the backend server. + + Returns: + str: The URL of the API endpoint. + """ return self.backend["url"]+self.backend["api"] - def set_endpoints(self, endpoints=None): - """Set the list of endpoints.""" + def set_endpoints(self, endpoints: Optional[list[dict]] = None) -> None: + """ + Set the list of endpoints used by this object. + + Args: + endpoints (list): A list of endpoint dictionaries. Each dictionary should contain information about an endpoint. + If None, resets the endpoints to their default value. + + Raises: + ValueError: If endpoints is not a list. + + Notes: + Endpoints can be reset to their default value by passing None as the argument. + """ if endpoints is not None: if not isinstance(endpoints, list): raise ValueError("Endpoints must be a list, even if there is just one model") self.endpoints = endpoints - def get_endpoints(self): + def get_endpoints(self) -> list[dict]: """ - Get the list of endpoints. + Get the complete list of endpoints. Returns: List of endpoints """ return self.endpoints + + def get_endpoints_with_key(self, key: str) -> list[dict]: + """ + Returns a list of endpoint dictionaries that contain the specified key. - def fetch_models(self): + Args: + key (str): The key to search for in the endpoint dictionaries. + + Returns: + List[Dict]: A list of endpoint dictionaries containing the specified key. + """ + return [ep for ep in self.endpoints if key in ep] + + + def fetch_models(self) -> None: """ Fetch models from endpoints and update the endpoint dictionaries. - Returns: None """ logger = logging.getLogger(__name__) for endpoint in self.endpoints: - if endpoint["provider"] == "ollama": - headers = { - "Content-Type": "application/json", - } - if "requestOptions" in endpoint: # Check if authentication is needed - headers.update({ - "Authorization": endpoint["requestOptions"]["headers"]["Authorization"] - }) + try: + if endpoint["provider"] == "ollama": + headers = { + "Content-Type": "application/json", + } + if "requestOptions" in endpoint: # Check if authentication is needed + headers.update({ + "Authorization": endpoint["requestOptions"]["headers"]["Authorization"] + }) - try: models_response = requests.get(endpoint["url"] + "/api/tags", headers=headers) models_response.raise_for_status() # Raise an exception for HTTP errors - models = models_response.json() - except requests.exceptions.RequestException as e: - logger.error("Error fetching models from backend: %s", str(e)) - continue - if isinstance(models, dict) and 'error' in models: - logger.error('Error fetching models from backend: %s', models['error']) - else: - endpoint["models"] = models.get("models", []) # Get the list of models directly + try: + models = models_response.json() + except json.JSONDecodeError as e: + logger.error(f"Failed to parse JSON response: {e}") + continue + + if isinstance(models, dict) and 'error' in models: # Unclear if requests to any API actually add this in the response + logger.error('Error fetching models from backend: %s', models['error']) + else: + endpoint["models"] = models.get("models", []) # Get the list of models directly + except requests.exceptions.RequestException as e: + logger.error(f"Request error: {e}") + except Exception as e: + logger.error(f"Unexpected error: {e}") + + return # No value returned - def get_list_of_available_llms(self, endpoint=None): - """Return a sorted list of LLMs available at endpoint""" + def get_list_of_available_llms(self, endpoint: Optional[dict] = None) -> Optional[list[str]]: + """ + Returns a sorted list of Large Language Models (LLMs) available at the specified endpoint. + + Args: + endpoint (dict): Optional endpoint dictionary to retrieve LLMs from. If not provided, will use internal endpoint configuration. + + Returns: + list: A sorted list of LLM names (strings). Returns None if no LLMs are found or endpoint is invalid. + """ llm_list = None if isinstance(endpoint["models"], list): llm_list = sorted([list_item['name'] for list_item in endpoint["models"]], key=str.lower) - return llm_list \ No newline at end of file + return llm_list + \ No newline at end of file