# [DEF:ConfigManagerModule:Module] # # @TIER: STANDARD # @SEMANTICS: config, manager, persistence, postgresql # @PURPOSE: Manages application configuration persisted in database with one-time migration from JSON. # @LAYER: Core # @RELATION: DEPENDS_ON -> ConfigModels # @RELATION: DEPENDS_ON -> AppConfigRecord # @RELATION: CALLS -> logger # # @INVARIANT: Configuration must always be valid according to AppConfig model. # @PUBLIC_API: ConfigManager # [SECTION: IMPORTS] import json import os from pathlib import Path from typing import Optional, List from sqlalchemy.orm import Session from .config_models import AppConfig, Environment, GlobalSettings, StorageConfig from .database import SessionLocal from ..models.config import AppConfigRecord from .logger import logger, configure_logger, belief_scope # [/SECTION] # [DEF:ConfigManager:Class] # @TIER: STANDARD # @PURPOSE: A class to handle application configuration persistence and management. class ConfigManager: # [DEF:__init__:Function] # @TIER: STANDARD # @PURPOSE: Initializes the ConfigManager. # @PRE: isinstance(config_path, str) and len(config_path) > 0 # @POST: self.config is an instance of AppConfig # @PARAM: config_path (str) - Path to legacy JSON config (used only for initial migration fallback). def __init__(self, config_path: str = "config.json"): with belief_scope("__init__"): assert isinstance(config_path, str) and config_path, "config_path must be a non-empty string" logger.info(f"[ConfigManager][Entry] Initializing with legacy path {config_path}") self.config_path = Path(config_path) self.config: AppConfig = self._load_config() configure_logger(self.config.settings.logging) assert isinstance(self.config, AppConfig), "self.config must be an instance of AppConfig" logger.info("[ConfigManager][Exit] Initialized") # [/DEF:__init__:Function] # [DEF:_default_config:Function] # @PURPOSE: Returns default application configuration. # @RETURN: AppConfig - Default configuration. def _default_config(self) -> AppConfig: return AppConfig( environments=[], settings=GlobalSettings(storage=StorageConfig()), ) # [/DEF:_default_config:Function] # [DEF:_load_from_legacy_file:Function] # @PURPOSE: Loads legacy configuration from config.json for migration fallback. # @RETURN: AppConfig - Loaded or default configuration. def _load_from_legacy_file(self) -> AppConfig: with belief_scope("_load_from_legacy_file"): if not self.config_path.exists(): logger.info("[_load_from_legacy_file][Action] Legacy config file not found, using defaults") return self._default_config() try: with open(self.config_path, "r", encoding="utf-8") as f: data = json.load(f) logger.info("[_load_from_legacy_file][Coherence:OK] Legacy configuration loaded") return AppConfig(**data) except Exception as e: logger.error(f"[_load_from_legacy_file][Coherence:Failed] Error loading legacy config: {e}") return self._default_config() # [/DEF:_load_from_legacy_file:Function] # [DEF:_get_record:Function] # @PURPOSE: Loads config record from DB. # @PARAM: session (Session) - DB session. # @RETURN: Optional[AppConfigRecord] - Existing record or None. def _get_record(self, session: Session) -> Optional[AppConfigRecord]: return session.query(AppConfigRecord).filter(AppConfigRecord.id == "global").first() # [/DEF:_get_record:Function] # [DEF:_load_config:Function] # @PURPOSE: Loads the configuration from DB or performs one-time migration from JSON file. # @PRE: DB session factory is available. # @POST: isinstance(return, AppConfig) # @RETURN: AppConfig - Loaded configuration. def _load_config(self) -> AppConfig: with belief_scope("_load_config"): session: Session = SessionLocal() try: record = self._get_record(session) if record and record.payload: logger.info("[_load_config][Coherence:OK] Configuration loaded from database") return AppConfig(**record.payload) logger.info("[_load_config][Action] No database config found, migrating legacy config") config = self._load_from_legacy_file() self._save_config_to_db(config, session=session) return config except Exception as e: logger.error(f"[_load_config][Coherence:Failed] Error loading config from DB: {e}") return self._default_config() finally: session.close() # [/DEF:_load_config:Function] # [DEF:_save_config_to_db:Function] # @PURPOSE: Saves the provided configuration object to DB. # @PRE: isinstance(config, AppConfig) # @POST: Configuration saved to database. # @PARAM: config (AppConfig) - The configuration to save. # @PARAM: session (Optional[Session]) - Existing DB session for transactional reuse. def _save_config_to_db(self, config: AppConfig, session: Optional[Session] = None): with belief_scope("_save_config_to_db"): assert isinstance(config, AppConfig), "config must be an instance of AppConfig" owns_session = session is None db = session or SessionLocal() try: record = self._get_record(db) payload = config.model_dump() if record is None: record = AppConfigRecord(id="global", payload=payload) db.add(record) else: record.payload = payload db.commit() logger.info("[_save_config_to_db][Action] Configuration saved to database") except Exception as e: db.rollback() logger.error(f"[_save_config_to_db][Coherence:Failed] Failed to save: {e}") raise finally: if owns_session: db.close() # [/DEF:_save_config_to_db:Function] # [DEF:save:Function] # @PURPOSE: Saves the current configuration state to DB. # @PRE: self.config is set. # @POST: self._save_config_to_db called. def save(self): with belief_scope("save"): self._save_config_to_db(self.config) # [/DEF:save:Function] # [DEF:get_config:Function] # @PURPOSE: Returns the current configuration. # @RETURN: AppConfig - The current configuration. def get_config(self) -> AppConfig: with belief_scope("get_config"): return self.config # [/DEF:get_config:Function] # [DEF:update_global_settings:Function] # @PURPOSE: Updates the global settings and persists the change. # @PRE: isinstance(settings, GlobalSettings) # @POST: self.config.settings updated and saved. # @PARAM: settings (GlobalSettings) - The new global settings. def update_global_settings(self, settings: GlobalSettings): with belief_scope("update_global_settings"): logger.info("[update_global_settings][Entry] Updating settings") assert isinstance(settings, GlobalSettings), "settings must be an instance of GlobalSettings" self.config.settings = settings self.save() configure_logger(settings.logging) logger.info("[update_global_settings][Exit] Settings updated") # [/DEF:update_global_settings:Function] # [DEF:validate_path:Function] # @PURPOSE: Validates if a path exists and is writable. # @PARAM: path (str) - The path to validate. # @RETURN: tuple (bool, str) - (is_valid, message) def validate_path(self, path: str) -> tuple[bool, str]: with belief_scope("validate_path"): p = os.path.abspath(path) if not os.path.exists(p): try: os.makedirs(p, exist_ok=True) except Exception as e: return False, f"Path does not exist and could not be created: {e}" if not os.access(p, os.W_OK): return False, "Path is not writable" return True, "Path is valid and writable" # [/DEF:validate_path:Function] # [DEF:get_environments:Function] # @PURPOSE: Returns the list of configured environments. # @RETURN: List[Environment] - List of environments. def get_environments(self) -> List[Environment]: with belief_scope("get_environments"): return self.config.environments # [/DEF:get_environments:Function] # [DEF:has_environments:Function] # @PURPOSE: Checks if at least one environment is configured. # @RETURN: bool - True if at least one environment exists. def has_environments(self) -> bool: with belief_scope("has_environments"): return len(self.config.environments) > 0 # [/DEF:has_environments:Function] # [DEF:get_environment:Function] # @PURPOSE: Returns a single environment by ID. # @PARAM: env_id (str) - The ID of the environment to retrieve. # @RETURN: Optional[Environment] - The environment with the given ID, or None. def get_environment(self, env_id: str) -> Optional[Environment]: with belief_scope("get_environment"): for env in self.config.environments: if env.id == env_id: return env return None # [/DEF:get_environment:Function] # [DEF:add_environment:Function] # @PURPOSE: Adds a new environment to the configuration. # @PARAM: env (Environment) - The environment to add. def add_environment(self, env: Environment): with belief_scope("add_environment"): logger.info(f"[add_environment][Entry] Adding environment {env.id}") assert isinstance(env, Environment), "env must be an instance of Environment" self.config.environments = [e for e in self.config.environments if e.id != env.id] self.config.environments.append(env) self.save() logger.info("[add_environment][Exit] Environment added") # [/DEF:add_environment:Function] # [DEF:update_environment:Function] # @PURPOSE: Updates an existing environment. # @PARAM: env_id (str) - The ID of the environment to update. # @PARAM: updated_env (Environment) - The updated environment data. # @RETURN: bool - True if updated, False otherwise. def update_environment(self, env_id: str, updated_env: Environment) -> bool: with belief_scope("update_environment"): logger.info(f"[update_environment][Entry] Updating {env_id}") assert env_id and isinstance(env_id, str), "env_id must be a non-empty string" assert isinstance(updated_env, Environment), "updated_env must be an instance of Environment" for i, env in enumerate(self.config.environments): if env.id == env_id: if updated_env.password == "********": updated_env.password = env.password self.config.environments[i] = updated_env self.save() logger.info(f"[update_environment][Coherence:OK] Updated {env_id}") return True logger.warning(f"[update_environment][Coherence:Failed] Environment {env_id} not found") return False # [/DEF:update_environment:Function] # [DEF:delete_environment:Function] # @PURPOSE: Deletes an environment by ID. # @PARAM: env_id (str) - The ID of the environment to delete. def delete_environment(self, env_id: str): with belief_scope("delete_environment"): logger.info(f"[delete_environment][Entry] Deleting {env_id}") assert env_id and isinstance(env_id, str), "env_id must be a non-empty string" original_count = len(self.config.environments) self.config.environments = [e for e in self.config.environments if e.id != env_id] if len(self.config.environments) < original_count: self.save() logger.info(f"[delete_environment][Action] Deleted {env_id}") else: logger.warning(f"[delete_environment][Coherence:Failed] Environment {env_id} not found") # [/DEF:delete_environment:Function] # [/DEF:ConfigManager:Class] # [/DEF:ConfigManagerModule:Module]