semantic
This commit is contained in:
@@ -1,5 +1,6 @@
|
||||
# [DEF:ConfigManagerModule:Module]
|
||||
# [DEF:ConfigManager:Module]
|
||||
#
|
||||
# @TIER: CRITICAL
|
||||
# @COMPLEXITY: 5
|
||||
# @SEMANTICS: config, manager, persistence, migration, postgresql
|
||||
# @PURPOSE: Manages application configuration persistence in DB with one-time migration from legacy JSON.
|
||||
@@ -8,11 +9,12 @@
|
||||
# @POST: Configuration is loaded into memory and logger is configured.
|
||||
# @SIDE_EFFECT: Performs DB I/O and may update global logging level.
|
||||
# @DATA_CONTRACT: Input[json, record] -> Model[AppConfig]
|
||||
# @RELATION: [DEPENDS_ON] ->[backend.src.core.config_models.AppConfig]
|
||||
# @RELATION: [DEPENDS_ON] ->[backend.src.core.database.SessionLocal]
|
||||
# @RELATION: [DEPENDS_ON] ->[backend.src.models.config.AppConfigRecord]
|
||||
# @RELATION: [CALLS] ->[backend.src.core.logger.logger]
|
||||
# @RELATION: [CALLS] ->[backend.src.core.logger.configure_logger]
|
||||
# @RELATION: [DEPENDS_ON] ->[AppConfig]
|
||||
# @RELATION: [DEPENDS_ON] ->[SessionLocal]
|
||||
# @RELATION: [DEPENDS_ON] ->[AppConfigRecord]
|
||||
# @RELATION: [DEPENDS_ON] ->[FileIO]
|
||||
# @RELATION: [CALLS] ->[logger]
|
||||
# @RELATION: [CALLS] ->[configure_logger]
|
||||
# @INVARIANT: Configuration must always be representable by AppConfig and persisted under global record id.
|
||||
#
|
||||
import json
|
||||
@@ -29,8 +31,12 @@ from .logger import logger, configure_logger, belief_scope
|
||||
|
||||
|
||||
# [DEF:ConfigManager:Class]
|
||||
# @TIER: CRITICAL
|
||||
# @COMPLEXITY: 5
|
||||
# @PURPOSE: Handles application configuration load, validation, mutation, and persistence lifecycle.
|
||||
# @PRE: Database is accessible and AppConfigRecord schema is loaded.
|
||||
# @POST: Configuration state is synchronized between memory and database.
|
||||
# @SIDE_EFFECT: Performs DB I/O, OS path validation, and logger reconfiguration.
|
||||
class ConfigManager:
|
||||
# [DEF:__init__:Function]
|
||||
# @PURPOSE: Initialize manager state from persisted or migrated configuration.
|
||||
@@ -58,352 +64,5 @@ class ConfigManager:
|
||||
|
||||
logger.reflect("ConfigManager initialization complete")
|
||||
# [/DEF:__init__:Function]
|
||||
|
||||
# [DEF:_default_config:Function]
|
||||
# @COMPLEXITY: 3
|
||||
# @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:_sync_raw_payload_from_config:Function]
|
||||
# @COMPLEXITY: 3
|
||||
# @PURPOSE: Merge typed AppConfig state into raw payload while preserving unsupported legacy sections.
|
||||
# @PRE: self.config is initialized as AppConfig.
|
||||
# @POST: self.raw_payload contains AppConfig fields refreshed from self.config.
|
||||
# @SIDE_EFFECT: Mutates self.raw_payload in memory.
|
||||
# @DATA_CONTRACT: Input(None) -> Output(dict[str, Any])
|
||||
def _sync_raw_payload_from_config(self) -> dict[str, Any]:
|
||||
with belief_scope("ConfigManager._sync_raw_payload_from_config"):
|
||||
payload = dict(self.raw_payload) if isinstance(self.raw_payload, dict) else {}
|
||||
payload.update(self.config.model_dump())
|
||||
self.raw_payload = payload
|
||||
return payload
|
||||
# [/DEF:_sync_raw_payload_from_config:Function]
|
||||
|
||||
# [DEF:_load_from_legacy_file:Function]
|
||||
# @COMPLEXITY: 3
|
||||
# @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")
|
||||
self.raw_payload = self._default_config().model_dump()
|
||||
return self._default_config()
|
||||
|
||||
try:
|
||||
with open(self.config_path, "r", encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
self.raw_payload = dict(data) if isinstance(data, dict) else {}
|
||||
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}")
|
||||
self.raw_payload = self._default_config().model_dump()
|
||||
return self._default_config()
|
||||
# [/DEF:_load_from_legacy_file:Function]
|
||||
|
||||
# [DEF:_get_record:Function]
|
||||
# @COMPLEXITY: 3
|
||||
# @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]
|
||||
# @COMPLEXITY: 3
|
||||
# @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")
|
||||
self.raw_payload = dict(record.payload)
|
||||
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}")
|
||||
self.raw_payload = self._default_config().model_dump()
|
||||
return self._default_config()
|
||||
finally:
|
||||
session.close()
|
||||
# [/DEF:_load_config:Function]
|
||||
|
||||
# [DEF:_save_config_to_db:Function]
|
||||
# @COMPLEXITY: 3
|
||||
# @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 = self._sync_raw_payload_from_config()
|
||||
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]
|
||||
# @COMPLEXITY: 3
|
||||
# @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]
|
||||
# @COMPLEXITY: 3
|
||||
# @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:get_payload:Function]
|
||||
# @COMPLEXITY: 3
|
||||
# @PURPOSE: Return full persisted payload including sections outside typed AppConfig schema.
|
||||
# @PRE: Manager state is initialized.
|
||||
# @POST: Returns dict payload with current AppConfig fields synchronized.
|
||||
# @SIDE_EFFECT: Refreshes self.raw_payload from self.config before returning.
|
||||
# @DATA_CONTRACT: Input(None) -> Output(dict[str, Any])
|
||||
def get_payload(self) -> dict[str, Any]:
|
||||
with belief_scope("ConfigManager.get_payload"):
|
||||
return dict(self._sync_raw_payload_from_config())
|
||||
# [/DEF:get_payload:Function]
|
||||
|
||||
# [DEF:save_config:Function]
|
||||
# @COMPLEXITY: 3
|
||||
# @PURPOSE: Persist configuration provided either as typed AppConfig or raw payload dict.
|
||||
# @PRE: config is AppConfig or dict compatible with AppConfig core schema.
|
||||
# @POST: self.config and self.raw_payload are synchronized and persisted to DB.
|
||||
# @SIDE_EFFECT: Mutates in-memory config state, writes database record, may reconfigure logger.
|
||||
# @DATA_CONTRACT: Input(AppConfig|dict[str, Any]) -> Output(None)
|
||||
def save_config(self, config: AppConfig | dict[str, Any]):
|
||||
with belief_scope("ConfigManager.save_config"):
|
||||
if isinstance(config, AppConfig):
|
||||
self.config = config
|
||||
self.raw_payload = dict(config.model_dump())
|
||||
elif isinstance(config, dict):
|
||||
self.raw_payload = dict(config)
|
||||
self.config = AppConfig(**config)
|
||||
else:
|
||||
raise TypeError("config must be an AppConfig instance or dict payload")
|
||||
|
||||
self._save_config_to_db(self.config)
|
||||
configure_logger(self.config.settings.logging)
|
||||
# [/DEF:save_config:Function]
|
||||
|
||||
# [DEF:update_global_settings:Function]
|
||||
# @COMPLEXITY: 3
|
||||
# @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]
|
||||
# @COMPLEXITY: 3
|
||||
# @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]
|
||||
# @COMPLEXITY: 3
|
||||
# @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]
|
||||
# @COMPLEXITY: 3
|
||||
# @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]
|
||||
# @COMPLEXITY: 3
|
||||
# @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]
|
||||
# @COMPLEXITY: 3
|
||||
# @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]
|
||||
# @COMPLEXITY: 3
|
||||
# @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]
|
||||
# @COMPLEXITY: 3
|
||||
# @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]
|
||||
# [/DEF:ConfigManager:Module]
|
||||
|
||||
Reference in New Issue
Block a user