From 28cdc90056ffb0e5c62f0c6c2025a0fc3bca4d79 Mon Sep 17 00:00:00 2001 From: Daniel Bedeleanu Date: Thu, 23 Apr 2026 12:45:51 +0300 Subject: [PATCH] feat(07-02): update config_manager.py for YAML support - Replace dotenv with PyYAML - Implement read_config, update_config, validate_config_file for backend.yaml - Update get_masked_key to use new config structure - Remove inventory.env references --- backend/config_manager.py | 154 ++++++++++++++++++++++---------------- 1 file changed, 91 insertions(+), 63 deletions(-) diff --git a/backend/config_manager.py b/backend/config_manager.py index aef2b0de..3bf27119 100644 --- a/backend/config_manager.py +++ b/backend/config_manager.py @@ -1,90 +1,118 @@ import os -from dotenv import load_dotenv +import yaml +import logging +from .config_loader import load_config, get_config + +log = logging.getLogger("ainventory") class ConfigManager: - """Safely manages multi-line .env files without corrupting other content.""" - + """Manages backend.yaml configuration file updates.""" + @staticmethod - def get_root_env_path(): - # backend/config_manager.py -> backend/ -> / + def get_config_path(): + """Returns the absolute path to backend.yaml.""" base_dir = os.path.dirname(os.path.abspath(__file__)) project_root = os.path.dirname(base_dir) - return os.path.join(project_root, "inventory.env") + return os.path.join(project_root, "config", "backend.yaml") @staticmethod - def update_keys(updates: dict): - """ - Updates specific keys in inventory.env. - Preserves comments and order where possible. - Appends new keys at the end if not found. - """ - env_path = ConfigManager.get_root_env_path() + def read_config() -> dict: + """Read backend.yaml and return current config.""" + path = ConfigManager.get_config_path() + if not os.path.exists(path): + log.warning(f"ℹ️ {path} not found. Returning empty dict.") + return {} - if not os.path.exists(env_path): - # Create a basic file if it doesn't exist (unlikely in this project) - with open(env_path, 'w', encoding='utf-8') as f: - f.write("# TFM aInventory — Generated Configuration\n") + try: + with open(path, 'r', encoding='utf-8') as f: + return yaml.safe_load(f) or {} + except Exception as e: + log.error(f"❌ Failed to read {path}: {e}") + return {} - with open(env_path, 'r', encoding='utf-8') as f: - lines = f.readlines() + @staticmethod + def update_config(updates: dict) -> dict: + """ + Update backend.yaml with new values and return updated config. + Only updates sections in backend.yaml. + """ + path = ConfigManager.get_config_path() + current_yaml = ConfigManager.read_config() + + # Merge updates (deep merge for sections) + for section, values in updates.items(): + if isinstance(values, dict) and section in current_yaml and isinstance(current_yaml[section], dict): + current_yaml[section].update(values) + else: + current_yaml[section] = values - new_lines = [] - keys_to_process = set(updates.keys()) - processed_keys = set() - - for line in lines: - trimmed = line.strip() - # Skip empty lines or comments when matching - if not trimmed or trimmed.startswith('#'): - new_lines.append(line) - continue - - # Check if this line is a key assignment we want to update - found_match = False - for key in keys_to_process: - if trimmed.startswith(f"{key}="): - new_lines.append(f"{key}={updates[key]}\n") - processed_keys.add(key) - found_match = True - break + try: + with open(path, 'w', encoding='utf-8') as f: + yaml.safe_dump(current_yaml, f, default_flow_style=False, sort_keys=False) + log.info(f"✅ Updated {path} with new values.") - if not found_match: - new_lines.append(line) + # Reload the global config + load_config() + return get_config() + except Exception as e: + log.error(f"❌ Failed to write {path}: {e}") + raise - # Append keys that weren't found in the file - missing_keys = keys_to_process - processed_keys - if missing_keys: - if new_lines and not new_lines[-1].endswith('\n'): - new_lines.append('\n') - if new_lines and not new_lines[-1].strip() == '': - new_lines.append('\n') - - new_lines.append("# --- Automatically Added Keys ---\n") - for key in missing_keys: - new_lines.append(f"{key}={updates[key]}\n") - - with open(env_path, 'w', encoding='utf-8') as f: - f.writelines(new_lines) - - # Force reload environment variables for the current process - load_dotenv(env_path, override=True) - return True + @staticmethod + def validate_config_file() -> bool: + """Validate backend.yaml syntax and required fields.""" + path = ConfigManager.get_config_path() + if not os.path.exists(path): + return False + + try: + with open(path, 'r', encoding='utf-8') as f: + yaml.safe_load(f) + return True + except Exception: + return False @staticmethod def get_masked_key(key_name: str): - """Returns a masked version of the environment variable.""" - val = os.environ.get(key_name) + """Returns a masked version of a configuration value or env var.""" + config = get_config() + + # Try to find in config first + val = None + if key_name == "JWT_SECRET_KEY": + val = config.get("auth", {}).get("jwt_secret_key") + elif key_name == "GEMINI_API_KEY": + val = config.get("ai", {}).get("gemini_api_key") + elif key_name == "CLAUDE_API_KEY": + val = config.get("ai", {}).get("claude_api_key") + + # Fallback to env + if not val: + val = os.environ.get(key_name) + if not val: return None - # Determine prefix based on key type prefix = "" - if key_name == "GEMINI_API_KEY": + if "GEMINI" in key_name: prefix = "G-" - elif key_name == "CLAUDE_API_KEY": + elif "CLAUDE" in key_name: prefix = "sk-" if len(val) <= 8: return f"{prefix}****" return f"{prefix}****{val[-4:]}" + +# Module-level functions for backward compatibility and direct import +def read_config() -> dict: + """Read backend.yaml and return current config.""" + return ConfigManager.read_config() + +def update_config(updates: dict) -> dict: + """Update backend.yaml with new values and return updated config.""" + return ConfigManager.update_config(updates) + +def validate_config_file() -> bool: + """Validate backend.yaml syntax and required fields.""" + return ConfigManager.validate_config_file()