Skip to content

Add ConfigManager for unified configuration management #239

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
AnkushMalaker wants to merge 1 commit into from
Closed

Add ConfigManager for unified configuration management #239

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
102 changes: 35 additions & 67 deletions backends/advanced/init.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,16 +28,27 @@ def __init__(self, args=None):
self.console = Console()
self.config: Dict[str, Any] = {}
self.args = args or argparse.Namespace()
self.config_yml_path = Path("../../config.yml") # Repo root config.yml
self.config_yml_data = None

# Check if we're in the right directory
if not Path("pyproject.toml").exists() or not Path("src").exists():
self.console.print("[red][ERROR][/red] Please run this script from the backends/advanced directory")
sys.exit(1)

# Load config.yml if it exists
self.load_config_yml()
# Initialize ConfigManager
repo_root = Path.cwd().parent.parent # backends/advanced -> repo root
if str(repo_root) not in sys.path:
sys.path.insert(0, str(repo_root))

from config_manager import ConfigManager

self.config_manager = ConfigManager(service_path="backends/advanced")
self.console.print(f"[blue][INFO][/blue] Using config.yml at: {self.config_manager.config_yml_path}")

# Load existing config or create default structure
self.config_yml_data = self.config_manager.get_full_config()
if not self.config_yml_data:
self.console.print("[yellow][WARNING][/yellow] config.yml not found, will create default structure")
self.config_yml_data = self._get_default_config_structure()
Comment on lines +37 to +51
Copy link
Contributor

@coderabbitai coderabbitai bot Jan 1, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion | 🟠 Major

Add error handling for ConfigManager import.

The dynamic sys.path manipulation and import could fail if the directory structure is unexpected or if config_manager.py is missing. Consider wrapping the import in a try-except block to provide a clear error message.

🔎 Proposed fix with error handling 
         # Initialize ConfigManager
         repo_root = Path.cwd().parent.parent  # backends/advanced -> repo root
         if str(repo_root) not in sys.path:
             sys.path.insert(0, str(repo_root))
 
-        from config_manager import ConfigManager
+        try:
+            from config_manager import ConfigManager
+        except ImportError as e:
+            self.console.print("[red][ERROR][/red] Failed to import ConfigManager. "
+                             "Ensure config_manager.py exists at repository root.")
+            raise RuntimeError("ConfigManager import failed") from e
 
         self.config_manager = ConfigManager(service_path="backends/advanced")
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
# Initialize ConfigManager
repo_root = Path.cwd().parent.parent # backends/advanced -> repo root
if str(repo_root) not in sys.path:
sys.path.insert(0, str(repo_root))
from config_manager import ConfigManager
self.config_manager = ConfigManager(service_path="backends/advanced")
self.console.print(f"[blue][INFO][/blue] Using config.yml at: {self.config_manager.config_yml_path}")
# Load existing config or create default structure
self.config_yml_data = self.config_manager.get_full_config()
if not self.config_yml_data:
self.console.print("[yellow][WARNING][/yellow] config.yml not found, will create default structure")
self.config_yml_data = self._get_default_config_structure()
# Initialize ConfigManager
repo_root = Path.cwd().parent.parent # backends/advanced -> repo root
if str(repo_root) not in sys.path:
sys.path.insert(0, str(repo_root))
try:
from config_manager import ConfigManager
except ImportError as e:
self.console.print("[red][ERROR][/red] Failed to import ConfigManager. "
"Ensure config_manager.py exists at repository root.")
raise RuntimeError("ConfigManager import failed") from e
self.config_manager = ConfigManager(service_path="backends/advanced")
self.console.print(f"[blue][INFO][/blue] Using config.yml at: {self.config_manager.config_yml_path}")
# Load existing config or create default structure
self.config_yml_data = self.config_manager.get_full_config()
if not self.config_yml_data:
self.console.print("[yellow][WARNING][/yellow] config.yml not found, will create default structure")
self.config_yml_data = self._get_default_config_structure()
🤖 Prompt for AI Agents 
In backends/advanced/init.py around lines 37 to 51, the dynamic sys.path
manipulation and "from config_manager import ConfigManager" can raise
ImportError or other exceptions if the file or path is missing; wrap the import
in a try-except block that catches ImportError/Exception, print a clear error to
self.console (including the exception message and the attempted repo_root path),
and then fail fast (raise a RuntimeError or call sys.exit(1)) so the failure is
explicit and actionable; keep the sys.path insertion before the import and
ensure any caught exception includes the original traceback or message in the
output.


def print_header(self, title: str):
"""Print a colorful header"""
Expand Down Expand Up @@ -126,21 +137,6 @@ def mask_api_key(self, key: str, show_chars: int = 5) -> str:

return f"{key_clean[:show_chars]}{'*' * min(15, len(key_clean) - show_chars * 2)}{key_clean[-show_chars:]}"

def load_config_yml(self):
"""Load config.yml from repository root"""
if not self.config_yml_path.exists():
self.console.print(f"[yellow][WARNING][/yellow] config.yml not found at {self.config_yml_path}")
self.console.print("[yellow]Will create a new config.yml during setup[/yellow]")
self.config_yml_data = self._get_default_config_structure()
return

try:
with open(self.config_yml_path, 'r') as f:
self.config_yml_data = yaml.safe_load(f)
self.console.print(f"[blue][INFO][/blue] Loaded existing config.yml")
except Exception as e:
self.console.print(f"[red][ERROR][/red] Failed to load config.yml: {e}")
self.config_yml_data = self._get_default_config_structure()

def _get_default_config_structure(self) -> Dict[str, Any]:
"""Return default config.yml structure if file doesn't exist"""
Expand All @@ -163,36 +159,6 @@ def _get_default_config_structure(self) -> Dict[str, Any]:
}
}

def save_config_yml(self):
"""Save config.yml back to repository root"""
try:
# Backup existing config.yml if it exists
if self.config_yml_path.exists():
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_path = self.config_yml_path.parent / f"config.yml.backup.{timestamp}"
shutil.copy2(self.config_yml_path, backup_path)
self.console.print(f"[blue][INFO][/blue] Backed up config.yml to {backup_path.name}")

# Write updated config
with open(self.config_yml_path, 'w') as f:
yaml.dump(self.config_yml_data, f, default_flow_style=False, sort_keys=False)

self.console.print("[green][SUCCESS][/green] config.yml updated successfully")
except Exception as e:
self.console.print(f"[red][ERROR][/red] Failed to save config.yml: {e}")
raise

def update_config_default(self, key: str, value: str):
"""Update a default value in config.yml"""
if "defaults" not in self.config_yml_data:
self.config_yml_data["defaults"] = {}
self.config_yml_data["defaults"][key] = value

def update_memory_config(self, updates: Dict[str, Any]):
"""Update memory configuration in config.yml"""
if "memory" not in self.config_yml_data:
self.config_yml_data["memory"] = {}
self.config_yml_data["memory"].update(updates)

def setup_authentication(self):
"""Configure authentication settings"""
Expand Down Expand Up @@ -306,8 +272,8 @@ def setup_llm(self):
if api_key:
self.config["OPENAI_API_KEY"] = api_key
# Update config.yml to use OpenAI models
self.update_config_default("llm", "openai-llm")
self.update_config_default("embedding", "openai-embed")
self.config_manager.update_config_defaults({"llm": "openai-llm", "embedding": "openai-embed"})
self.config_yml_data = self.config_manager.get_full_config() # Reload to stay in sync
self.console.print("[green][SUCCESS][/green] OpenAI configured in config.yml")
self.console.print("[blue][INFO][/blue] Set defaults.llm: openai-llm")
self.console.print("[blue][INFO][/blue] Set defaults.embedding: openai-embed")
Expand All @@ -317,8 +283,8 @@ def setup_llm(self):
elif choice == "2":
self.console.print("[blue][INFO][/blue] Ollama selected")
# Update config.yml to use Ollama models
self.update_config_default("llm", "local-llm")
self.update_config_default("embedding", "local-embed")
self.config_manager.update_config_defaults({"llm": "local-llm", "embedding": "local-embed"})
self.config_yml_data = self.config_manager.get_full_config() # Reload to stay in sync
self.console.print("[green][SUCCESS][/green] Ollama configured in config.yml")
self.console.print("[blue][INFO][/blue] Set defaults.llm: local-llm")
self.console.print("[blue][INFO][/blue] Set defaults.embedding: local-embed")
Expand All @@ -327,7 +293,8 @@ def setup_llm(self):
elif choice == "3":
self.console.print("[blue][INFO][/blue] Skipping LLM setup - memory extraction disabled")
# Disable memory extraction in config.yml
self.update_memory_config({"extraction": {"enabled": False}})
self.config_manager.update_memory_config({"extraction": {"enabled": False}})
self.config_yml_data = self.config_manager.get_full_config() # Reload to stay in sync

def setup_memory(self):
"""Configure memory provider - updates config.yml"""
Expand All @@ -347,9 +314,10 @@ def setup_memory(self):
qdrant_url = self.prompt_value("Qdrant URL", "qdrant")
self.config["QDRANT_BASE_URL"] = qdrant_url

# Update config.yml
self.update_memory_config({"provider": "chronicle"})
self.console.print("[green][SUCCESS][/green] Chronicle memory provider configured in config.yml")
# Update config.yml (also updates .env automatically)
self.config_manager.update_memory_config({"provider": "chronicle"})
self.config_yml_data = self.config_manager.get_full_config() # Reload to stay in sync
self.console.print("[green][SUCCESS][/green] Chronicle memory provider configured in config.yml and .env")

elif choice == "2":
self.console.print("[blue][INFO][/blue] OpenMemory MCP selected")
Expand All @@ -359,8 +327,8 @@ def setup_memory(self):
user_id = self.prompt_value("OpenMemory user ID", "openmemory")
timeout = self.prompt_value("OpenMemory timeout (seconds)", "30")

# Update config.yml with OpenMemory MCP settings
self.update_memory_config({
# Update config.yml with OpenMemory MCP settings (also updates .env automatically)
self.config_manager.update_memory_config({
"provider": "openmemory_mcp",
"openmemory_mcp": {
"server_url": mcp_url,
Expand All @@ -369,7 +337,8 @@ def setup_memory(self):
"timeout": int(timeout)
}
})
self.console.print("[green][SUCCESS][/green] OpenMemory MCP configured in config.yml")
self.config_yml_data = self.config_manager.get_full_config() # Reload to stay in sync
self.console.print("[green][SUCCESS][/green] OpenMemory MCP configured in config.yml and .env")
self.console.print("[yellow][WARNING][/yellow] Remember to start OpenMemory: cd ../../extras/openmemory-mcp && docker compose up -d")

elif choice == "3":
Expand All @@ -378,15 +347,16 @@ def setup_memory(self):
mycelia_url = self.prompt_value("Mycelia API URL", "http://localhost:5173")
timeout = self.prompt_value("Mycelia timeout (seconds)", "30")

# Update config.yml with Mycelia settings
self.update_memory_config({
# Update config.yml with Mycelia settings (also updates .env automatically)
self.config_manager.update_memory_config({
"provider": "mycelia",
"mycelia": {
"api_url": mycelia_url,
"timeout": int(timeout)
}
})
self.console.print("[green][SUCCESS][/green] Mycelia memory provider configured in config.yml")
self.config_yml_data = self.config_manager.get_full_config() # Reload to stay in sync
self.console.print("[green][SUCCESS][/green] Mycelia memory provider configured in config.yml and .env")
self.console.print("[yellow][WARNING][/yellow] Make sure Mycelia is running at the configured URL")

def setup_optional_services(self):
Expand Down Expand Up @@ -559,10 +529,8 @@ def generate_env_file(self):

self.console.print("[green][SUCCESS][/green] .env file configured successfully with secure permissions")

# Save config.yml with all updates
self.console.print()
self.console.print("[blue][INFO][/blue] Saving configuration to config.yml...")
self.save_config_yml()
# Note: config.yml is automatically saved by ConfigManager when updates are made
self.console.print("[blue][INFO][/blue] Configuration saved to config.yml and .env (via ConfigManager)")

def copy_config_templates(self):
"""Copy other configuration files"""
Expand Down
Loading
Loading