# [DEF:ConfigManagerModule:Module] # # @TIER: CRITICAL # @SEMANTICS: config, manager, persistence, migration, postgresql # @PURPOSE: Manages application configuration persistence in DB with one-time migration from legacy JSON. # @LAYER: Domain # @RELATION: [DEPENDS_ON] ->[ConfigModels] # @RELATION: [DEPENDS_ON] ->[SessionLocal] # @RELATION: [DEPENDS_ON] ->[AppConfigRecord] # @RELATION: [CALLS] ->[logger] # @RELATION: [CALLS] ->[configure_logger] # @RELATION: [BINDS_TO] ->[ConfigManager] # @INVARIANT: Configuration must always be representable by AppConfig and persisted under global record id. # 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 # [DEF:ConfigManager:Class] # @TIER: CRITICAL # @PURPOSE: Handles application configuration load, validation, mutation, and persistence lifecycle. class ConfigManager: # [DEF:__init__:Function] # @PURPOSE: Initialize manager state from persisted or migrated configuration. # @PRE: config_path is a non-empty string path. # @POST: self.config is initialized as AppConfig and logger is configured. # @SIDE_EFFECT: Reads config sources and updates logging configuration. # @DATA_CONTRACT: Input(str config_path) -> Output(None; self.config: AppConfig) def __init__(self, config_path: str = "config.json"): with belief_scope("ConfigManager.__init__"): if not isinstance(config_path, str) or not config_path: logger.explore("Invalid config_path provided", extra={"path": config_path}) raise ValueError("config_path must be a non-empty string") logger.reason(f"Initializing ConfigManager with legacy path: {config_path}") self.config_path = Path(config_path) self.config: AppConfig = self._load_config() configure_logger(self.config.settings.logging) if not isinstance(self.config, AppConfig): logger.explore("Config loading resulted in invalid type", extra={"type": type(self.config)}) raise TypeError("self.config must be an instance of AppConfig") logger.reflect("ConfigManager initialization complete") # [/DEF:__init__:Function] # [DEF:_default_config:Function] # @PURPOSE: Build default application configuration fallback. # @PRE: None. # @POST: Returns valid AppConfig with empty environments and default storage settings. # @SIDE_EFFECT: None. # @DATA_CONTRACT: Input(None) -> Output(AppConfig) def _default_config(self) -> AppConfig: with belief_scope("_default_config"): return AppConfig( environments=[], settings=GlobalSettings(storage=StorageConfig()), ) # [/DEF:_default_config:Function] # [DEF:_load_from_legacy_file:Function] # @PURPOSE: Load legacy JSON configuration for migration fallback path. # @PRE: self.config_path is initialized. # @POST: Returns AppConfig from file payload or safe default. # @SIDE_EFFECT: Filesystem read and error logging. # @DATA_CONTRACT: Input(Path self.config_path) -> Output(AppConfig) 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: Resolve global configuration record from DB. # @PRE: session is an active SQLAlchemy Session. # @POST: Returns record when present, otherwise None. # @SIDE_EFFECT: Database read query. # @DATA_CONTRACT: Input(Session) -> Output(Optional[AppConfigRecord]) def _get_record(self, session: Session) -> Optional[AppConfigRecord]: with belief_scope("_get_record"): return session.query(AppConfigRecord).filter(AppConfigRecord.id == "global").first() # [/DEF:_get_record:Function] # [DEF:_load_config:Function] # @PURPOSE: Load configuration from DB or perform one-time migration from legacy JSON. # @PRE: SessionLocal factory is available and AppConfigRecord schema is accessible. # @POST: Returns valid AppConfig and closes opened DB session. # @SIDE_EFFECT: Database read/write, possible migration write, logging. # @DATA_CONTRACT: Input(None) -> Output(AppConfig) def _load_config(self) -> AppConfig: with belief_scope("ConfigManager._load_config"): session: Session = SessionLocal() try: record = self._get_record(session) if record and record.payload: logger.reason("Configuration found in database") config = AppConfig(**record.payload) logger.reflect("Database configuration validated") return config logger.reason("No database config found, initiating legacy migration") config = self._load_from_legacy_file() self._save_config_to_db(config, session=session) logger.reflect("Legacy configuration migrated to database") return config except Exception as e: logger.explore(f"Error loading config from DB: {e}") return self._default_config() finally: session.close() # [/DEF:_load_config:Function] # [DEF:_save_config_to_db:Function] # @PURPOSE: Persist provided AppConfig into the global DB configuration record. # @PRE: config is AppConfig; session is either None or an active Session. # @POST: Global DB record payload equals config.model_dump() when commit succeeds. # @SIDE_EFFECT: Database insert/update, commit/rollback, logging. # @DATA_CONTRACT: Input(AppConfig, Optional[Session]) -> Output(None) def _save_config_to_db(self, config: AppConfig, session: Optional[Session] = None): with belief_scope("ConfigManager._save_config_to_db"): if not isinstance(config, AppConfig): raise TypeError("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: logger.reason("Creating new global configuration record") record = AppConfigRecord(id="global", payload=payload) db.add(record) else: logger.reason("Updating existing global configuration record") record.payload = payload db.commit() logger.reflect("Configuration successfully committed to database") except Exception as e: db.rollback() logger.explore(f"Failed to save configuration: {e}") raise finally: if owns_session: db.close() # [/DEF:_save_config_to_db:Function] # [DEF:save:Function] # @PURPOSE: Persist current in-memory configuration state. # @PRE: self.config is initialized. # @POST: Current self.config is written to DB global record. # @SIDE_EFFECT: Database write and logging via delegated persistence call. # @DATA_CONTRACT: Input(None; self.config: AppConfig) -> Output(None) def save(self): with belief_scope("save"): self._save_config_to_db(self.config) # [/DEF:save:Function] # [DEF:get_config:Function] # @PURPOSE: Return current in-memory configuration snapshot. # @PRE: self.config is initialized. # @POST: Returns AppConfig reference stored in manager. # @SIDE_EFFECT: None. # @DATA_CONTRACT: Input(None) -> Output(AppConfig) def get_config(self) -> AppConfig: with belief_scope("get_config"): return self.config # [/DEF:get_config:Function] # [DEF:update_global_settings:Function] # @PURPOSE: Replace global settings and persist the resulting configuration. # @PRE: settings is GlobalSettings. # @POST: self.config.settings equals provided settings and DB state is updated. # @SIDE_EFFECT: Mutates self.config, DB write, logger reconfiguration, logging. # @DATA_CONTRACT: Input(GlobalSettings) -> Output(None) def update_global_settings(self, settings: GlobalSettings): with belief_scope("ConfigManager.update_global_settings"): if not isinstance(settings, GlobalSettings): raise TypeError("settings must be an instance of GlobalSettings") logger.reason("Updating global settings and persisting") self.config.settings = settings self.save() configure_logger(settings.logging) logger.reflect("Global settings updated and logger reconfigured") # [/DEF:update_global_settings:Function] # [DEF:validate_path:Function] # @PURPOSE: Validate that path exists and is writable, creating it when absent. # @PRE: path is a string path candidate. # @POST: Returns (True, msg) for writable path, else (False, reason). # @SIDE_EFFECT: Filesystem directory creation attempt and OS permission checks. # @DATA_CONTRACT: Input(str path) -> Output(tuple[bool, str]) 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: Return all configured environments. # @PRE: self.config is initialized. # @POST: Returns list of Environment models from current configuration. # @SIDE_EFFECT: None. # @DATA_CONTRACT: Input(None) -> Output(List[Environment]) def get_environments(self) -> List[Environment]: with belief_scope("get_environments"): return self.config.environments # [/DEF:get_environments:Function] # [DEF:has_environments:Function] # @PURPOSE: Check whether at least one environment exists in configuration. # @PRE: self.config is initialized. # @POST: Returns True iff environment list length is greater than zero. # @SIDE_EFFECT: None. # @DATA_CONTRACT: Input(None) -> Output(bool) 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: Resolve a configured environment by identifier. # @PRE: env_id is string identifier. # @POST: Returns matching Environment when found; otherwise None. # @SIDE_EFFECT: None. # @DATA_CONTRACT: Input(str env_id) -> Output(Optional[Environment]) 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: Upsert environment by id into configuration and persist. # @PRE: env is Environment. # @POST: Configuration contains provided env id with new payload persisted. # @SIDE_EFFECT: Mutates environment list, DB write, logging. # @DATA_CONTRACT: Input(Environment) -> Output(None) def add_environment(self, env: Environment): with belief_scope("ConfigManager.add_environment"): if not isinstance(env, Environment): raise TypeError("env must be an instance of Environment") logger.reason(f"Adding/Updating environment: {env.id}") self.config.environments = [e for e in self.config.environments if e.id != env.id] self.config.environments.append(env) self.save() logger.reflect(f"Environment {env.id} persisted") # [/DEF:add_environment:Function] # [DEF:update_environment:Function] # @PURPOSE: Update existing environment by id and preserve masked password placeholder behavior. # @PRE: env_id is non-empty string and updated_env is Environment. # @POST: Returns True and persists update when target exists; else returns False. # @SIDE_EFFECT: May mutate environment list, DB write, logging. # @DATA_CONTRACT: Input(str env_id, Environment updated_env) -> Output(bool) def update_environment(self, env_id: str, updated_env: Environment) -> bool: with belief_scope("ConfigManager.update_environment"): if not env_id or not isinstance(env_id, str): raise ValueError("env_id must be a non-empty string") if not isinstance(updated_env, Environment): raise TypeError("updated_env must be an instance of Environment") logger.reason(f"Attempting to update environment: {env_id}") for i, env in enumerate(self.config.environments): if env.id == env_id: if updated_env.password == "********": logger.reason("Preserving existing password for masked update") updated_env.password = env.password self.config.environments[i] = updated_env self.save() logger.reflect(f"Environment {env_id} updated and saved") return True logger.explore(f"Environment {env_id} not found for update") return False # [/DEF:update_environment:Function] # [DEF:delete_environment:Function] # @PURPOSE: Delete environment by id and persist when deletion occurs. # @PRE: env_id is non-empty string. # @POST: Environment is removed when present; otherwise configuration is unchanged. # @SIDE_EFFECT: May mutate environment list, conditional DB write, logging. # @DATA_CONTRACT: Input(str env_id) -> Output(None) def delete_environment(self, env_id: str): with belief_scope("ConfigManager.delete_environment"): if not env_id or not isinstance(env_id, str): raise ValueError("env_id must be a non-empty string") logger.reason(f"Attempting to delete environment: {env_id}") 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.reflect(f"Environment {env_id} deleted and configuration saved") else: logger.explore(f"Environment {env_id} not found for deletion") # [/DEF:delete_environment:Function] # [/DEF:ConfigManager:Class] # [/DEF:ConfigManagerModule:Module]