|
""" |
|
Configuration utilities for GAIA implementation. |
|
|
|
This module provides functions for loading, validating, and accessing configuration |
|
settings from various sources like environment variables, files, and defaults. |
|
""" |
|
|
|
import os |
|
import json |
|
import logging |
|
from typing import Dict, Any, Optional, List, Union |
|
from pathlib import Path |
|
|
|
from src.gaia.utils.common import get_nested_value, set_nested_value, load_json_file |
|
|
|
logger = logging.getLogger("gaia_agent.utils.config_utils") |
|
|
|
def load_from_env(prefix: str = "GAIA_") -> Dict[str, Any]: |
|
""" |
|
Load configuration from environment variables. |
|
|
|
Args: |
|
prefix: Prefix for environment variables to load (e.g., 'GAIA_') |
|
|
|
Returns: |
|
Dictionary with environment variable values |
|
""" |
|
config = {} |
|
|
|
for key, value in os.environ.items(): |
|
if key.startswith(prefix): |
|
|
|
config_key = key[len(prefix):].lower() |
|
|
|
|
|
if '_' in config_key: |
|
parts = config_key.split('_') |
|
current = config |
|
|
|
|
|
for part in parts[:-1]: |
|
if part not in current: |
|
current[part] = {} |
|
current = current[part] |
|
|
|
|
|
current[parts[-1]] = value |
|
else: |
|
config[config_key] = value |
|
|
|
return config |
|
|
|
def load_from_file(file_path: str) -> Optional[Dict[str, Any]]: |
|
""" |
|
Load configuration from a JSON file. |
|
|
|
Args: |
|
file_path: Path to the configuration file |
|
|
|
Returns: |
|
Dictionary with configuration values, or None if loading failed |
|
""" |
|
return load_json_file(file_path) |
|
|
|
def load_from_env_file(env_path: Union[str, Path]) -> Dict[str, str]: |
|
""" |
|
Load environment variables from a .env file. |
|
|
|
Args: |
|
env_path: Path to the .env file |
|
|
|
Returns: |
|
Dictionary with environment variables |
|
""" |
|
env_vars = {} |
|
|
|
if not os.path.exists(env_path): |
|
return env_vars |
|
|
|
try: |
|
with open(env_path, 'r') as f: |
|
for line in f: |
|
line = line.strip() |
|
if not line or line.startswith('#'): |
|
continue |
|
|
|
|
|
if line.startswith('export '): |
|
line = line[7:] |
|
|
|
if '=' in line: |
|
key, value = line.split('=', 1) |
|
|
|
value = value.strip('\'"') |
|
env_vars[key.strip()] = value |
|
except Exception as e: |
|
logger.error(f"Error loading .env file {env_path}: {str(e)}") |
|
|
|
return env_vars |
|
|
|
def merge_configs(base: Dict[str, Any], override: Dict[str, Any]) -> Dict[str, Any]: |
|
""" |
|
Merge two configuration dictionaries, with override values taking precedence. |
|
|
|
Args: |
|
base: Base configuration dictionary |
|
override: Override configuration dictionary |
|
|
|
Returns: |
|
Merged configuration dictionary |
|
""" |
|
result = base.copy() |
|
|
|
for key, value in override.items(): |
|
if isinstance(value, dict) and key in result and isinstance(result[key], dict): |
|
|
|
result[key] = merge_configs(result[key], value) |
|
else: |
|
|
|
result[key] = value |
|
|
|
return result |
|
|
|
def find_config_file(file_name: str, search_paths: List[str]) -> Optional[str]: |
|
""" |
|
Find a configuration file in a list of search paths. |
|
|
|
Args: |
|
file_name: Name of the configuration file |
|
search_paths: List of paths to search |
|
|
|
Returns: |
|
Full path to the configuration file, or None if not found |
|
""" |
|
for path in search_paths: |
|
full_path = os.path.join(path, file_name) |
|
if os.path.exists(full_path): |
|
return full_path |
|
|
|
return None |
|
|
|
def get_config_value(config: Dict[str, Any], key: str, default: Any = None, env_var: Optional[str] = None) -> Any: |
|
""" |
|
Get a configuration value from a configuration dictionary, with fallback to environment variable. |
|
|
|
Args: |
|
config: Configuration dictionary |
|
key: Configuration key (can use dot notation for nested keys) |
|
default: Default value if key is not found |
|
env_var: Environment variable to check if key is not found in config |
|
|
|
Returns: |
|
Configuration value or default |
|
""" |
|
|
|
value = get_nested_value(config, key, None) |
|
if value is not None: |
|
return value |
|
|
|
|
|
if env_var and env_var in os.environ: |
|
return os.environ[env_var] |
|
|
|
|
|
return default |
|
|
|
def convert_value_type(value: str, target_type: type) -> Any: |
|
""" |
|
Convert a string value to a specified type. |
|
|
|
Args: |
|
value: String value to convert |
|
target_type: Target type to convert to |
|
|
|
Returns: |
|
Converted value |
|
""" |
|
if target_type == bool: |
|
return value.lower() in ('true', 'yes', 'y', '1', 'on') |
|
elif target_type == int: |
|
return int(value) |
|
elif target_type == float: |
|
return float(value) |
|
elif target_type == list: |
|
return [item.strip() for item in value.split(',')] |
|
else: |
|
return value |