96 lines
2.7 KiB
Python
96 lines
2.7 KiB
Python
"""
|
|
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
|
|
|
|
_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 _KEYRING_OK and password:
|
|
keyring.set_password(_SERVICE, profile_id, password)
|
|
|
|
|
|
def load_password(profile_id: str) -> str:
|
|
if _KEYRING_OK:
|
|
try:
|
|
return keyring.get_password(_SERVICE, profile_id) or ""
|
|
except Exception:
|
|
return ""
|
|
return ""
|
|
|
|
|
|
def delete_password(profile_id: str) -> None:
|
|
if _KEYRING_OK:
|
|
try:
|
|
keyring.delete_password(_SERVICE, profile_id)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
# ── 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)
|