busya
b5e741077d
Core changes: - Add @defgroup/@ingroup to 1791 C2+ contracts (555 files) for HCA 128× pre-training DSA grouping - Add §0.1 Pre-Training Frequency matrix to semantics-core - Add §VIII Attention Architecture rules (ATTN_1-4) with MLA/CSA/HCA/DSA mechanics - Add @defgroup/@ingroup to canonical syntax (§II) and all contract examples Agent prompts (5 files): - Add ZERO-STATE RATIONALE with MLA/CSA/HCA/DSA compression mechanics - Add pre-training note: @RATIONALE/@REJECTED are in-context learned tags - svelte-coder: add missing #region contract, fix Svelte rule violations - python-coder/fullstack-coder: honor function contracts from speckit plan - qa-tester: add attention compliance audit (P3 ATTN_1-4 checks) Skills (6 files): - Translate all axiom_config descriptions to English - Fix doc_dirs to index .opencode/ and .specify/ - Deduplicate 5× complexity_rules → single global_tags catalog - Reduce semantics-svelte 591→485 lines (remove duplicate code blocks) - Fix semantics-testing: 'Short IDs' → 'Short hierarchical IDs' - Fix all examples: flat IDs → hierarchical Domain.Name format - Fix Svelte examples: replace raw Tailwind + <button> with semantic tokens + /ui Speckit workflow (commands + templates): - speckit.plan: add Function-Level Contracts for C3+ with @PRE/@POST/@TEST_EDGE - speckit.plan: add Attention Compliance Gate (ATTN_1-4 before contract generation) - speckit.tasks: add function contract inlining format (constraints in task description) - speckit.specify: load semantics-core for spec density rules - spec-template: add #region contract, @SEMANTICS grouping, hierarchical IDs - ux-reference-template: add #region wrapper - plan-template: add attention gate, @defgroup/@ingroup guidance - tasks-template: add attention audit + rebuild + orphan check tasks - constitution.md: translate to English, add Principle VIII (attention-optimized contracts) Reference modules rewritten (hierarchical IDs + full contracts): - Auth.Jwt: 6 child contracts with @RATIONALE/@REJECTED/@TEST_EDGE - Api.Auth: 5 endpoints with @TEST_EDGE + molecular CoT markers - Migration.Model: @defgroup Migration with 18 @ACTION + 6 @INVARIANT Scripts: - add_defgroup_ingroup.py: zero-risk additive @ingroup migration (1791 insertions) - migrate_hierarchical.py: flat→hierarchical ID dry-run analysis (792 contracts) - merge_prompts.py: merge all prompts/skills/commands into one review file Config: - axiom_config.yaml: 749→395 lines (-47%), English, doc_dirs include prompts - Fix test_datasets.py import collision (rename → test_datasets_routes.py) - Fix test_preview.py: SupersetClient→get_superset_client, AsyncMock, logger f-string
655 lines
31 KiB
Python
655 lines
31 KiB
Python
# #region ConfigManager [C:5] [TYPE Module] [SEMANTICS sqlalchemy, validate, migration, config-manager]
|
|
# @defgroup Core Module group.
|
|
#
|
|
# @BRIEF Manages application configuration persistence in DB with one-time migration from legacy JSON.
|
|
# @LAYER Domain
|
|
# @PRE Database schema for AppConfigRecord must be initialized.
|
|
# @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]
|
|
# @INVARIANT Configuration must always be representable by AppConfig and persisted under global record id.
|
|
# @INVARIANT Environment passwords are encrypted at rest using Fernet (see [SEC:C-1]).
|
|
# @RELATION DEPENDS_ON -> [AppConfig]
|
|
# @RELATION DEPENDS_ON -> [SessionLocal]
|
|
# @RELATION DEPENDS_ON -> [AppConfigRecord]
|
|
# @RELATION DEPENDS_ON -> [EncryptionManager]
|
|
# @RELATION CALLS -> [logger]
|
|
# @RELATION CALLS -> [configure_logger]
|
|
#
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..models.config import AppConfigRecord
|
|
from ..models.mapping import Environment as EnvironmentRecord
|
|
from .config_models import AppConfig, Environment, GlobalSettings
|
|
from .database import SessionLocal
|
|
from .encryption import EncryptionManager
|
|
from .logger import belief_scope, configure_logger, logger
|
|
|
|
|
|
# #region ConfigManager [C:5] [TYPE Class]
|
|
# @defgroup Core Module group.
|
|
# @BRIEF 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:
|
|
# #region __init__ [TYPE 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.raw_payload: dict[str, Any] = {}
|
|
self.config: AppConfig = self._load_config()
|
|
configure_logger(self.config.settings.logging)
|
|
# Disable propagation on app loggers to prevent duplicate output
|
|
# via uvicorn's root logger handler (see ADR LOG-002 in logger.py).
|
|
# We do this caller-side so tests calling configure_logger directly
|
|
# retain propagation for caplog capture.
|
|
from .cot_logger import cot_logger # noqa: F811
|
|
from .logger import logger as _app_logger
|
|
_app_logger.propagate = False # type: ignore[assignment]
|
|
cot_logger.propagate = False
|
|
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")
|
|
# #endregion __init__
|
|
|
|
# #region _get_encryption_manager [TYPE Function]
|
|
# @PURPOSE: Lazy-init EncryptionManager for password encryption/decryption.
|
|
# @POST Returns EncryptionManager or None if ENCRYPTION_KEY is unavailable.
|
|
# @SIDE_EFFECT May log a warning if key is missing.
|
|
# @RATIONALE Encryption manager cannot be initialized before ensure_encryption_key()
|
|
# is called at startup. This lazy init allows early bootstrap paths
|
|
# (legacy migration, test setup) to work without encryption.
|
|
def _get_encryption_manager(self) -> EncryptionManager | None:
|
|
if not hasattr(self, "_encryption"):
|
|
try:
|
|
self._encryption = EncryptionManager()
|
|
except (RuntimeError, Exception) as exc:
|
|
logger.reason(
|
|
"EncryptionManager unavailable; passwords stored in plaintext. "
|
|
"Ensure ENCRYPTION_KEY is set for production.",
|
|
extra={"error": str(exc)},
|
|
)
|
|
self._encryption = None
|
|
return self._encryption
|
|
# #endregion _get_encryption_manager
|
|
|
|
# #region _encrypt_env_passwords [TYPE Function]
|
|
# @PURPOSE: Encrypt all environment passwords in the raw payload dict before DB persistence.
|
|
# @PRE payload["environments"] is a list of dicts with optional "password" keys.
|
|
# @POST Password fields are encrypted in-place (skips masked/empty/already-encrypted).
|
|
# @SIDE_EFFECT Reads ENCRYPTION_KEY; logs warnings on failure.
|
|
def _encrypt_env_passwords(self, payload: dict) -> None:
|
|
encryption = self._get_encryption_manager()
|
|
if encryption is None:
|
|
return
|
|
environments = payload.get("environments", [])
|
|
if not isinstance(environments, list):
|
|
return
|
|
for env in environments:
|
|
pwd = env.get("password", "")
|
|
if not pwd or pwd == "********":
|
|
continue
|
|
# Skip if already encrypted (safe-decrypt check)
|
|
try:
|
|
encryption.decrypt(pwd)
|
|
continue
|
|
except Exception:
|
|
pass
|
|
try:
|
|
env["password"] = encryption.encrypt(pwd)
|
|
except Exception as exc:
|
|
logger.explore(
|
|
"Failed to encrypt environment password",
|
|
extra={"error": str(exc)},
|
|
)
|
|
# #endregion _encrypt_env_passwords
|
|
|
|
# #region _decrypt_env_passwords [TYPE Function]
|
|
# @PURPOSE: Decrypt all environment passwords in the AppConfig after loading from DB.
|
|
# @PRE config.environments is populated from DB payload.
|
|
# @POST Password fields are decrypted in-place (skips plaintext legacy values).
|
|
def _decrypt_env_passwords(self, config: AppConfig) -> None:
|
|
encryption = self._get_encryption_manager()
|
|
if encryption is None:
|
|
return
|
|
for env in config.environments:
|
|
if not env.password:
|
|
continue
|
|
try:
|
|
env.password = encryption.decrypt(env.password)
|
|
except Exception:
|
|
pass # Assume plaintext (legacy data or test fixture)
|
|
# #endregion _decrypt_env_passwords
|
|
|
|
# #region _apply_features_from_env [TYPE Function]
|
|
# @PURPOSE: Read FEATURES__* env vars and apply them to a GlobalSettings features config.
|
|
# @SIDE_EFFECT Reads os.environ; mutates settings.features in-place.
|
|
# @RATIONALE Env vars seed the initial defaults. After first bootstrap, DB is source of truth.
|
|
@staticmethod
|
|
def _apply_features_from_env(settings: GlobalSettings) -> None:
|
|
with belief_scope("ConfigManager._apply_features_from_env"):
|
|
dataset_review_env = os.getenv("FEATURES__DATASET_REVIEW")
|
|
if dataset_review_env is not None:
|
|
parsed = dataset_review_env.strip().lower() in ("true", "1", "yes")
|
|
settings.features.dataset_review = parsed
|
|
logger.reason(
|
|
"Applied FEATURES__DATASET_REVIEW from env",
|
|
extra={"value": parsed, "raw": dataset_review_env},
|
|
)
|
|
health_monitor_env = os.getenv("FEATURES__HEALTH_MONITOR")
|
|
if health_monitor_env is not None:
|
|
parsed = health_monitor_env.strip().lower() in ("true", "1", "yes")
|
|
settings.features.health_monitor = parsed
|
|
logger.reason(
|
|
"Applied FEATURES__HEALTH_MONITOR from env",
|
|
extra={"value": parsed, "raw": health_monitor_env},
|
|
)
|
|
|
|
app_timezone_env = os.getenv("APP_TIMEZONE")
|
|
if app_timezone_env is not None:
|
|
settings.app_timezone = app_timezone_env.strip()
|
|
logger.reason(
|
|
"Applied APP_TIMEZONE from env",
|
|
extra={"value": settings.app_timezone},
|
|
)
|
|
# #endregion _apply_features_from_env
|
|
# #region _default_config [TYPE Function]
|
|
# @PURPOSE: Build default application configuration fallback.
|
|
def _default_config(self) -> AppConfig:
|
|
with belief_scope("ConfigManager._default_config"):
|
|
logger.reason("Building default AppConfig fallback")
|
|
config = AppConfig(environments=[], settings=GlobalSettings())
|
|
self._apply_features_from_env(config.settings)
|
|
return config
|
|
# #endregion _default_config
|
|
# #region _sync_raw_payload_from_config [TYPE Function]
|
|
# @PURPOSE: Merge typed AppConfig state into raw payload while preserving unsupported legacy sections.
|
|
def _sync_raw_payload_from_config(self) -> dict[str, Any]:
|
|
with belief_scope("ConfigManager._sync_raw_payload_from_config"):
|
|
typed_payload = self.config.model_dump()
|
|
merged_payload = dict(self.raw_payload or {})
|
|
merged_payload["environments"] = typed_payload.get("environments", [])
|
|
merged_payload["settings"] = typed_payload.get("settings", {})
|
|
self.raw_payload = merged_payload
|
|
logger.reason(
|
|
"Synchronized raw payload from typed config",
|
|
extra={
|
|
"environments_count": len(
|
|
merged_payload.get("environments", []) or []
|
|
),
|
|
"has_settings": "settings" in merged_payload,
|
|
"extra_sections": sorted(
|
|
key
|
|
for key in merged_payload
|
|
if key not in {"environments", "settings"}
|
|
),
|
|
},
|
|
)
|
|
return merged_payload
|
|
# #endregion _sync_raw_payload_from_config
|
|
# #region _load_from_legacy_file [TYPE Function]
|
|
# @PURPOSE: Load legacy JSON configuration for migration fallback path.
|
|
def _load_from_legacy_file(self) -> dict[str, Any]:
|
|
with belief_scope("ConfigManager._load_from_legacy_file"):
|
|
if not self.config_path.exists():
|
|
logger.reason(
|
|
"Legacy config file not found; using default payload",
|
|
extra={"path": str(self.config_path)},
|
|
)
|
|
return {}
|
|
logger.reason(
|
|
"Loading legacy config file", extra={"path": str(self.config_path)}
|
|
)
|
|
with self.config_path.open("r", encoding="utf-8") as fh:
|
|
payload = json.load(fh)
|
|
if not isinstance(payload, dict):
|
|
logger.explore(
|
|
"Legacy config payload is not a JSON object",
|
|
extra={
|
|
"path": str(self.config_path),
|
|
"type": type(payload).__name__,
|
|
},
|
|
)
|
|
raise ValueError("Legacy config payload must be a JSON object")
|
|
logger.reason(
|
|
"Legacy config file loaded successfully",
|
|
extra={"path": str(self.config_path), "keys": sorted(payload.keys())},
|
|
)
|
|
return payload
|
|
# #endregion _load_from_legacy_file
|
|
# #region _get_record [TYPE Function]
|
|
# @PURPOSE: Resolve global configuration record from DB.
|
|
def _get_record(self, session: Session) -> AppConfigRecord | None:
|
|
with belief_scope("ConfigManager._get_record"):
|
|
record = (
|
|
session.query(AppConfigRecord)
|
|
.filter(AppConfigRecord.id == "global")
|
|
.first()
|
|
)
|
|
logger.reason(
|
|
"Resolved app config record", extra={"exists": record is not None}
|
|
)
|
|
return record
|
|
# #endregion _get_record
|
|
# #region _load_config [TYPE Function]
|
|
# @PURPOSE: Load configuration from DB or perform one-time migration from legacy JSON.
|
|
def _load_config(self) -> AppConfig:
|
|
with belief_scope("ConfigManager._load_config"):
|
|
session = SessionLocal()
|
|
try:
|
|
record = self._get_record(session)
|
|
if record and isinstance(record.payload, dict):
|
|
logger.reason(
|
|
"Loading configuration from database",
|
|
extra={"record_id": record.id},
|
|
)
|
|
self.raw_payload = dict(record.payload)
|
|
config = AppConfig.model_validate(
|
|
{
|
|
"environments": self.raw_payload.get("environments", []),
|
|
"settings": self.raw_payload.get("settings", {}),
|
|
}
|
|
)
|
|
# Decrypt environment passwords after loading from DB (see [SEC:C-1])
|
|
self._decrypt_env_passwords(config)
|
|
self._sync_environment_records(session, config)
|
|
session.commit()
|
|
logger.reason(
|
|
"Database configuration validated successfully",
|
|
extra={
|
|
"environments_count": len(config.environments),
|
|
"payload_keys": sorted(self.raw_payload.keys()),
|
|
},
|
|
)
|
|
return config
|
|
logger.reason(
|
|
"Database configuration record missing; attempting legacy file migration",
|
|
extra={"legacy_path": str(self.config_path)},
|
|
)
|
|
legacy_payload = self._load_from_legacy_file()
|
|
if legacy_payload:
|
|
self.raw_payload = dict(legacy_payload)
|
|
config = AppConfig.model_validate(
|
|
{
|
|
"environments": self.raw_payload.get("environments", []),
|
|
"settings": self.raw_payload.get("settings", {}),
|
|
}
|
|
)
|
|
# Legacy config is plaintext — decrypt is a no-op but safe
|
|
self._decrypt_env_passwords(config)
|
|
self._apply_features_from_env(config.settings)
|
|
logger.reason(
|
|
"Legacy payload validated; persisting migrated configuration to database",
|
|
extra={
|
|
"environments_count": len(config.environments),
|
|
"payload_keys": sorted(self.raw_payload.keys()),
|
|
},
|
|
)
|
|
self._save_config_to_db(config, session=session)
|
|
return config
|
|
logger.reason(
|
|
"No persisted config found; falling back to default configuration"
|
|
)
|
|
config = self._default_config()
|
|
self.raw_payload = config.model_dump()
|
|
self._save_config_to_db(config, session=session)
|
|
return config
|
|
except (json.JSONDecodeError, TypeError, ValueError) as exc:
|
|
logger.explore(
|
|
"Recoverable config load failure; falling back to default configuration",
|
|
extra={"error": str(exc), "legacy_path": str(self.config_path)},
|
|
)
|
|
config = self._default_config()
|
|
self.raw_payload = config.model_dump()
|
|
return config
|
|
except Exception as exc:
|
|
logger.explore(
|
|
"Critical config load failure; re-raising persistence or validation error",
|
|
extra={"error": str(exc)},
|
|
)
|
|
raise
|
|
finally:
|
|
session.close()
|
|
# #endregion _load_config
|
|
# #region _sync_environment_records [TYPE Function]
|
|
# @PURPOSE: Mirror configured environments into the relational environments table used by FK-backed domain models.
|
|
def _sync_environment_records(self, session: Session, config: AppConfig) -> None:
|
|
with belief_scope("ConfigManager._sync_environment_records"):
|
|
configured_envs = list(config.environments or [])
|
|
persisted_records = session.query(EnvironmentRecord).all()
|
|
persisted_by_id = {
|
|
str(record.id or "").strip(): record for record in persisted_records
|
|
}
|
|
for environment in configured_envs:
|
|
normalized_id = str(environment.id or "").strip()
|
|
if not normalized_id:
|
|
continue
|
|
display_name = (
|
|
str(environment.name or normalized_id).strip() or normalized_id
|
|
)
|
|
normalized_url = str(environment.url or "").strip()
|
|
credentials_id = (
|
|
str(environment.username or "").strip() or normalized_id
|
|
)
|
|
record = persisted_by_id.get(normalized_id)
|
|
if record is None:
|
|
logger.reason(
|
|
"Creating relational environment record from typed config",
|
|
extra={
|
|
"environment_id": normalized_id,
|
|
"environment_name": display_name,
|
|
},
|
|
)
|
|
session.add(
|
|
EnvironmentRecord(
|
|
id=normalized_id,
|
|
name=display_name,
|
|
url=normalized_url,
|
|
credentials_id=credentials_id,
|
|
)
|
|
)
|
|
continue
|
|
record.name = display_name
|
|
record.url = normalized_url
|
|
record.credentials_id = credentials_id
|
|
# #endregion _sync_environment_records
|
|
# #region _delete_stale_environment_records [TYPE Function]
|
|
# @PURPOSE: Remove persisted environment records that are no longer in the configured environments.
|
|
# @PRE _sync_environment_records must already have run so the query returns a current view.
|
|
# @POST Stale Environment rows are deleted via the session (caller must commit).
|
|
# @SIDE_EFFECT Only safe to call during explicit save (_save_config_to_db), NOT during _load_config,
|
|
# because FK-dependent rows (task_records, database_mappings, etc.) may reference deleted rows.
|
|
def _delete_stale_environment_records(
|
|
self, session: Session, config: AppConfig
|
|
) -> None:
|
|
with belief_scope("ConfigManager._delete_stale_environment_records"):
|
|
persisted_records = session.query(EnvironmentRecord).all()
|
|
configured_ids = {
|
|
str(env.id or "").strip()
|
|
for env in (config.environments or [])
|
|
if str(env.id or "").strip()
|
|
}
|
|
for record in persisted_records:
|
|
record_id = str(record.id or "").strip()
|
|
if record_id and record_id not in configured_ids:
|
|
logger.reason(
|
|
"Deleting stale environment record",
|
|
extra={"environment_id": record_id},
|
|
)
|
|
session.delete(record)
|
|
# #endregion _delete_stale_environment_records
|
|
# #region _save_config_to_db [TYPE Function]
|
|
# @PURPOSE: Persist provided AppConfig into the global DB configuration record.
|
|
# @SIDE_EFFECT Encrypts environment passwords for at-rest security (see [SEC:C-1]);
|
|
# restores plaintext after save so in-memory config stays usable.
|
|
def _save_config_to_db(
|
|
self, config: AppConfig, session: Session | None = None
|
|
) -> None:
|
|
with belief_scope("ConfigManager._save_config_to_db"):
|
|
owns_session = session is None
|
|
db = session or SessionLocal()
|
|
try:
|
|
self.config = config
|
|
payload = self._sync_raw_payload_from_config()
|
|
# Encrypt passwords for at-rest security before DB write
|
|
self._encrypt_env_passwords(payload)
|
|
record = self._get_record(db)
|
|
if record is None:
|
|
logger.reason("Creating new global app config record")
|
|
record = AppConfigRecord(id="global", payload=payload)
|
|
db.add(record)
|
|
else:
|
|
logger.reason(
|
|
"Updating existing global app config record",
|
|
extra={"record_id": record.id},
|
|
)
|
|
record.payload = payload
|
|
self._sync_environment_records(db, config)
|
|
self._delete_stale_environment_records(db, config)
|
|
db.commit()
|
|
logger.reason(
|
|
"Configuration persisted to database",
|
|
extra={
|
|
"environments_count": len(
|
|
payload.get("environments", []) or []
|
|
),
|
|
"payload_keys": sorted(payload.keys()),
|
|
},
|
|
)
|
|
# Restore plaintext passwords in config after save
|
|
self._decrypt_env_passwords(config)
|
|
except Exception:
|
|
db.rollback()
|
|
logger.explore("Database save failed; transaction rolled back")
|
|
raise
|
|
finally:
|
|
if owns_session:
|
|
db.close()
|
|
# #endregion _save_config_to_db
|
|
# #region save [TYPE Function]
|
|
# @ingroup Core
|
|
# @PURPOSE: Persist current in-memory configuration state.
|
|
def save(self) -> None:
|
|
with belief_scope("ConfigManager.save"):
|
|
logger.reason("Persisting current in-memory configuration")
|
|
self._save_config_to_db(self.config)
|
|
# #endregion save
|
|
# #region get_config [TYPE Function]
|
|
# @ingroup Core
|
|
# @PURPOSE: Return current in-memory configuration snapshot.
|
|
def get_config(self) -> AppConfig:
|
|
with belief_scope("ConfigManager.get_config"):
|
|
return self.config
|
|
# #endregion get_config
|
|
# #region get_payload [TYPE Function]
|
|
# @ingroup Core
|
|
# @PURPOSE: Return full persisted payload including sections outside typed AppConfig schema.
|
|
def get_payload(self) -> dict[str, Any]:
|
|
with belief_scope("ConfigManager.get_payload"):
|
|
return self._sync_raw_payload_from_config()
|
|
# #endregion get_payload
|
|
# #region save_config [TYPE Function]
|
|
# @ingroup Core
|
|
# @PURPOSE: Persist configuration provided either as typed AppConfig or raw payload dict.
|
|
def save_config(self, config: Any) -> AppConfig:
|
|
with belief_scope("ConfigManager.save_config"):
|
|
if isinstance(config, AppConfig):
|
|
logger.reason("Saving typed AppConfig payload")
|
|
self.config = config
|
|
self.raw_payload = config.model_dump()
|
|
self._save_config_to_db(config)
|
|
return self.config
|
|
if isinstance(config, dict):
|
|
logger.reason(
|
|
"Saving raw config payload",
|
|
extra={"keys": sorted(config.keys())},
|
|
)
|
|
self.raw_payload = dict(config)
|
|
typed_config = AppConfig.model_validate(
|
|
{
|
|
"environments": self.raw_payload.get("environments", []),
|
|
"settings": self.raw_payload.get("settings", {}),
|
|
}
|
|
)
|
|
self.config = typed_config
|
|
self._save_config_to_db(typed_config)
|
|
return self.config
|
|
logger.explore(
|
|
"Unsupported config type supplied to save_config",
|
|
extra={"type": type(config).__name__},
|
|
)
|
|
raise TypeError("config must be AppConfig or dict")
|
|
# #endregion save_config
|
|
# #region update_global_settings [TYPE Function]
|
|
# @ingroup Core
|
|
# @PURPOSE: Replace global settings and persist the resulting configuration.
|
|
def update_global_settings(self, settings: GlobalSettings) -> AppConfig:
|
|
with belief_scope("ConfigManager.update_global_settings"):
|
|
logger.reason("Updating global settings")
|
|
self.config.settings = settings
|
|
self.save()
|
|
return self.config
|
|
# #endregion update_global_settings
|
|
# #region validate_path [TYPE Function]
|
|
# @ingroup Core
|
|
# @PURPOSE: Validate that path exists and is writable, creating it when absent.
|
|
def validate_path(self, path: str) -> tuple[bool, str]:
|
|
with belief_scope("ConfigManager.validate_path", f"path={path}"):
|
|
try:
|
|
target = Path(path).expanduser()
|
|
target.mkdir(parents=True, exist_ok=True)
|
|
if not target.exists():
|
|
return False, f"Path does not exist: {target}"
|
|
if not target.is_dir():
|
|
return False, f"Path is not a directory: {target}"
|
|
test_file = target / ".write_test"
|
|
with test_file.open("w", encoding="utf-8") as fh:
|
|
fh.write("ok")
|
|
test_file.unlink(missing_ok=True)
|
|
logger.reason("Path validation succeeded", extra={"path": str(target)})
|
|
return True, "OK"
|
|
except Exception as exc:
|
|
logger.explore(
|
|
"Path validation failed", extra={"path": path, "error": str(exc)}
|
|
)
|
|
return False, str(exc)
|
|
# #endregion validate_path
|
|
# #region get_environments [TYPE Function]
|
|
# @ingroup Core
|
|
# @PURPOSE: Return all configured environments.
|
|
def get_environments(self) -> list[Environment]:
|
|
with belief_scope("ConfigManager.get_environments"):
|
|
return list(self.config.environments)
|
|
# #endregion get_environments
|
|
# #region has_environments [TYPE Function]
|
|
# @ingroup Core
|
|
# @PURPOSE: Check whether at least one environment exists in configuration.
|
|
def has_environments(self) -> bool:
|
|
with belief_scope("ConfigManager.has_environments"):
|
|
return len(self.config.environments) > 0
|
|
# #endregion has_environments
|
|
# #region get_environment [TYPE Function]
|
|
# @ingroup Core
|
|
# @PURPOSE: Resolve a configured environment by identifier.
|
|
def get_environment(self, env_id: str) -> Environment | None:
|
|
with belief_scope("ConfigManager.get_environment", f"env_id={env_id}"):
|
|
normalized = str(env_id or "").strip()
|
|
if not normalized:
|
|
return None
|
|
for env in self.config.environments:
|
|
if env.id == normalized or env.name == normalized:
|
|
return env
|
|
return None
|
|
# #endregion get_environment
|
|
# #region add_environment [TYPE Function]
|
|
# @ingroup Core
|
|
# @PURPOSE: Upsert environment by id into configuration and persist.
|
|
def add_environment(self, env: Environment) -> AppConfig:
|
|
with belief_scope("ConfigManager.add_environment", f"env_id={env.id}"):
|
|
existing_index = next(
|
|
(
|
|
i
|
|
for i, item in enumerate(self.config.environments)
|
|
if item.id == env.id
|
|
),
|
|
None,
|
|
)
|
|
if env.is_default:
|
|
for item in self.config.environments:
|
|
item.is_default = False
|
|
if existing_index is None:
|
|
logger.reason("Appending new environment", extra={"env_id": env.id})
|
|
self.config.environments.append(env)
|
|
else:
|
|
logger.reason(
|
|
"Replacing existing environment during add",
|
|
extra={"env_id": env.id},
|
|
)
|
|
self.config.environments[existing_index] = env
|
|
if len(self.config.environments) == 1 and not any(
|
|
item.is_default for item in self.config.environments
|
|
):
|
|
self.config.environments[0].is_default = True
|
|
self.save()
|
|
return self.config
|
|
# #endregion add_environment
|
|
# #region update_environment [TYPE Function]
|
|
# @ingroup Core
|
|
# @PURPOSE: Update existing environment by id and preserve masked password placeholder behavior.
|
|
def update_environment(self, env_id: str, env: Environment) -> bool:
|
|
with belief_scope("ConfigManager.update_environment", f"env_id={env_id}"):
|
|
for index, existing in enumerate(self.config.environments):
|
|
if existing.id != env_id:
|
|
continue
|
|
update_data = env.model_dump()
|
|
if update_data.get("password") == "********":
|
|
update_data["password"] = existing.password
|
|
updated = Environment.model_validate(update_data)
|
|
if updated.is_default:
|
|
for item in self.config.environments:
|
|
item.is_default = False
|
|
elif existing.is_default and not updated.is_default:
|
|
updated.is_default = True
|
|
self.config.environments[index] = updated
|
|
logger.reason("Environment updated", extra={"env_id": env_id})
|
|
self.save()
|
|
return True
|
|
logger.explore(
|
|
"Environment update skipped; env not found", extra={"env_id": env_id}
|
|
)
|
|
return False
|
|
# #endregion update_environment
|
|
# #region delete_environment [TYPE Function]
|
|
# @ingroup Core
|
|
# @PURPOSE: Delete environment by id and persist when deletion occurs.
|
|
def delete_environment(self, env_id: str) -> bool:
|
|
with belief_scope("ConfigManager.delete_environment", f"env_id={env_id}"):
|
|
before = len(self.config.environments)
|
|
removed = [env for env in self.config.environments if env.id == env_id]
|
|
self.config.environments = [
|
|
env for env in self.config.environments if env.id != env_id
|
|
]
|
|
if len(self.config.environments) == before:
|
|
logger.explore(
|
|
"Environment delete skipped; env not found",
|
|
extra={"env_id": env_id},
|
|
)
|
|
return False
|
|
if removed and removed[0].is_default and self.config.environments:
|
|
self.config.environments[0].is_default = True
|
|
if self.config.settings.default_environment_id == env_id:
|
|
replacement = next(
|
|
(env.id for env in self.config.environments if env.is_default), None
|
|
)
|
|
self.config.settings.default_environment_id = replacement
|
|
logger.reason(
|
|
"Environment deleted",
|
|
extra={"env_id": env_id, "remaining": len(self.config.environments)},
|
|
)
|
|
self.save()
|
|
return True
|
|
# #endregion delete_environment
|
|
# #endregion ConfigManager
|
|
# #endregion ConfigManager
|