""" Connection profile persistence. Profiles are stored as JSON in ~/.dbclient/connections.json. Passwords are stored separately in the OS keychain via keyring. """ import json from pathlib import Path from typing import Optional from app.models.connection_model import ConnectionProfile from app.utils.logger import get_logger _log = get_logger(__name__) _CONN_FILE = Path.home() / ".dbclient" / "connections.json" try: import keyring _KEYRING_OK = True except ImportError: _KEYRING_OK = False _SERVICE = "DBClient" # ── Keyring helpers ────────────────────────────────────────────────────────── def save_password(profile_id: str, password: str) -> None: if not _KEYRING_OK: return if password: try: keyring.set_password(_SERVICE, profile_id, password) except Exception: _log.warning("keyring: failed to save password for %s", profile_id, exc_info=True) else: delete_password(profile_id) def load_password(profile_id: str) -> str: if _KEYRING_OK: try: return keyring.get_password(_SERVICE, profile_id) or "" except Exception: _log.warning("keyring: failed to load password for %s", profile_id, exc_info=True) return "" return "" def delete_password(profile_id: str) -> None: if _KEYRING_OK: try: keyring.delete_password(_SERVICE, profile_id) except Exception: pass # Already absent is fine # ── Profile CRUD ───────────────────────────────────────────────────────────── def _load_raw() -> list: if _CONN_FILE.exists(): try: with open(_CONN_FILE, "r", encoding="utf-8") as f: return json.load(f) except Exception: return [] return [] def _save_raw(data: list) -> None: _CONN_FILE.parent.mkdir(parents=True, exist_ok=True) with open(_CONN_FILE, "w", encoding="utf-8") as f: json.dump(data, f, indent=2) def load_profiles() -> list: """Return list of ConnectionProfile with passwords injected from keyring.""" profiles = [] for raw in _load_raw(): p = ConnectionProfile.from_dict(raw) p.password = load_password(p.id) profiles.append(p) return profiles def save_profile(profile: ConnectionProfile) -> None: """Upsert a profile (save or update).""" raw_list = _load_raw() # Replace existing or append found = False for i, raw in enumerate(raw_list): if raw.get("id") == profile.id: raw_list[i] = profile.to_dict() found = True break if not found: raw_list.append(profile.to_dict()) _save_raw(raw_list) save_password(profile.id, profile.password) def delete_profile(profile_id: str) -> None: """Remove a profile by ID.""" raw_list = [r for r in _load_raw() if r.get("id") != profile_id] _save_raw(raw_list) delete_password(profile_id)