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
This commit is contained in:
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user