danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Migrate config storage to SQLite database

Browse source 
ConfigManager now stores configuration in a SQLite database instead of a JSON file. The class supports migration from existing config.json files and falls back to defaults if no configuration is found. This change improves reliability and centralizes configuration management.
This commit is contained in:
Broque Thomas 2025-12-04 11:08:38 -08:00
parent bfabd26469
commit 36418e05b4
1 changed files with 220 additions and 32 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

252
config/settings.py
View file

@ -1,5 +1,6 @@
import json
import os
import sqlite3
from typing import Dict, Any, Optional
from cryptography.fernet import Fernet
from pathlib import Path
@ -9,8 +10,9 @@ class ConfigManager:
self.config_path = Path(config_path)
self.config_data: Dict[str, Any] = {}
self.encryption_key: Optional[bytes] = None
self.database_path = Path("database/music_library.db") # Hardcoded - same as MusicDatabase
self._load_config()
def _get_encryption_key(self) -> bytes:
key_file = self.config_path.parent / ".encryption_key"
if key_file.exists():
@ -22,48 +24,234 @@ class ConfigManager:
f.write(key)
key_file.chmod(0o600)
return key
def _ensure_database_exists(self):
"""Ensure database file and metadata table exist"""
try:
# Create database directory if it doesn't exist
self.database_path.parent.mkdir(parents=True, exist_ok=True)
# Connect to database (creates file if it doesn't exist)
conn = sqlite3.connect(str(self.database_path))
cursor = conn.cursor()
# Create metadata table if it doesn't exist
cursor.execute("""
CREATE TABLE IF NOT EXISTS metadata (
key TEXT PRIMARY KEY,
value TEXT,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.commit()
conn.close()
except Exception as e:
print(f"Warning: Could not ensure database exists: {e}")
def _load_from_database(self) -> Optional[Dict[str, Any]]:
"""Load configuration from database"""
try:
self._ensure_database_exists()
conn = sqlite3.connect(str(self.database_path))
cursor = conn.cursor()
cursor.execute("SELECT value FROM metadata WHERE key = 'app_config'")
row = cursor.fetchone()
conn.close()
if row and row[0]:
config_data = json.loads(row[0])
print("[OK] Configuration loaded from database")
return config_data
else:
return None
except Exception as e:
print(f"Warning: Could not load config from database: {e}")
return None
def _save_to_database(self, config_data: Dict[str, Any]) -> bool:
"""Save configuration to database"""
try:
self._ensure_database_exists()
conn = sqlite3.connect(str(self.database_path))
cursor = conn.cursor()
config_json = json.dumps(config_data, indent=2)
cursor.execute("""
INSERT OR REPLACE INTO metadata (key, value, updated_at)
VALUES ('app_config', ?, CURRENT_TIMESTAMP)
""", (config_json,))
conn.commit()
conn.close()
return True
except Exception as e:
print(f"Error: Could not save config to database: {e}")
return False
def _load_from_config_file(self) -> Optional[Dict[str, Any]]:
"""Load configuration from config.json file (for migration)"""
try:
if self.config_path.exists():
with open(self.config_path, 'r') as f:
config_data = json.load(f)
print(f"[OK] Configuration loaded from {self.config_path}")
return config_data
else:
return None
except Exception as e:
print(f"Warning: Could not load config from file: {e}")
return None
def _get_default_config(self) -> Dict[str, Any]:
"""Get default configuration"""
return {
"active_media_server": "plex",
"spotify": {
"client_id": "",
"client_secret": "",
"redirect_uri": "http://127.0.0.1:8888/callback"
},
"tidal": {
"client_id": "",
"client_secret": "",
"redirect_uri": "http://127.0.0.1:8889/tidal/callback"
},
"plex": {
"base_url": "",
"token": "",
"auto_detect": True
},
"jellyfin": {
"base_url": "",
"api_key": "",
"auto_detect": True
},
"navidrome": {
"base_url": "",
"username": "",
"password": "",
"auto_detect": True
},
"soulseek": {
"slskd_url": "",
"api_key": "",
"download_path": "./downloads",
"transfer_path": "./Transfer"
},
"listenbrainz": {
"token": ""
},
"logging": {
"path": "logs/app.log",
"level": "INFO"
},
"database": {
"path": "database/music_library.db",
"max_workers": 5
},
"metadata_enhancement": {
"enabled": True,
"embed_album_art": True
},
"playlist_sync": {
"create_backup": True
},
"settings": {
"audio_quality": "flac"
}
}
def _load_config(self):
if not self.config_path.exists():
raise FileNotFoundError(f"Configuration file not found: {self.config_path}")
with open(self.config_path, 'r') as f:
self.config_data = json.load(f)
"""
Load configuration with priority:
1. Database (primary storage)
2. config.json (migration from file-based config)
3. Defaults (fresh install)
"""
# Try loading from database first
config_data = self._load_from_database()
if config_data:
# Configuration exists in database
self.config_data = config_data
return
# Database is empty - try migration from config.json
config_data = self._load_from_config_file()
if config_data:
# Migrate from config.json to database
print("[MIGRATE] Migrating configuration from config.json to database...")
if self._save_to_database(config_data):
print("[OK] Configuration migrated successfully")
self.config_data = config_data
return
else:
print("[WARN] Migration failed - using file-based config")
self.config_data = config_data
return
# No config.json either - use defaults
print("[INFO] No existing configuration found - using defaults")
config_data = self._get_default_config()
# Try to save defaults to database
if self._save_to_database(config_data):
print("[OK] Default configuration saved to database")
else:
print("[WARN] Could not save defaults to database - using in-memory config")
self.config_data = config_data
def _save_config(self):
with open(self.config_path, 'w') as f:
json.dump(self.config_data, f, indent=2)
"""Save configuration to database"""
success = self._save_to_database(self.config_data)
if not success:
# Fallback: Try to save to config.json if database fails
print("[WARN] Database save failed - attempting file fallback")
try:
self.config_path.parent.mkdir(parents=True, exist_ok=True)
with open(self.config_path, 'w') as f:
json.dump(self.config_data, f, indent=2)
print("[OK] Configuration saved to config.json as fallback")
except Exception as e:
print(f"[ERROR] Failed to save configuration: {e}")
def get(self, key: str, default: Any = None) -> Any:
keys = key.split('.')
value = self.config_data
for k in keys:
if isinstance(value, dict) and k in value:
value = value[k]
else:
return default
return value
def set(self, key: str, value: Any):
keys = key.split('.')
config = self.config_data
for k in keys[:-1]:
if k not in config:
config[k] = {}
config = config[k]
config[keys[-1]] = value
self._save_config()
def get_spotify_config(self) -> Dict[str, str]:
return self.get('spotify', {})
def get_plex_config(self) -> Dict[str, str]:
return self.get('plex', {})
def get_jellyfin_config(self) -> Dict[str, str]:
return self.get('jellyfin', {})
@ -72,25 +260,25 @@ class ConfigManager:
def get_soulseek_config(self) -> Dict[str, str]:
return self.get('soulseek', {})
def get_settings(self) -> Dict[str, Any]:
return self.get('settings', {})
def get_database_config(self) -> Dict[str, str]:
return self.get('database', {})
def get_logging_config(self) -> Dict[str, str]:
return self.get('logging', {})
def get_active_media_server(self) -> str:
return self.get('active_media_server', 'plex')
def set_active_media_server(self, server: str):
"""Set the active media server (plex, jellyfin, or navidrome)"""
if server not in ['plex', 'jellyfin', 'navidrome']:
raise ValueError(f"Invalid media server: {server}")
self.set('active_media_server', server)
def get_active_media_server_config(self) -> Dict[str, str]:
"""Get configuration for the currently active media server"""
active_server = self.get_active_media_server()
@ -102,12 +290,12 @@ class ConfigManager:
return self.get_navidrome_config()
else:
return {}
def is_configured(self) -> bool:
spotify = self.get_spotify_config()
active_server = self.get_active_media_server()
soulseek = self.get_soulseek_config()
# Check active media server configuration
media_server_configured = False
if active_server == 'plex':
@ -119,22 +307,22 @@ class ConfigManager:
elif active_server == 'navidrome':
navidrome = self.get_navidrome_config()
media_server_configured = bool(navidrome.get('base_url')) and bool(navidrome.get('username')) and bool(navidrome.get('password'))
return (
bool(spotify.get('client_id')) and
bool(spotify.get('client_secret')) and
media_server_configured and
bool(soulseek.get('slskd_url'))
)
def validate_config(self) -> Dict[str, bool]:
active_server = self.get_active_media_server()
validation = {
'spotify': bool(self.get('spotify.client_id')) and bool(self.get('spotify.client_secret')),
'soulseek': bool(self.get('soulseek.slskd_url'))
}
# Validate all server types but mark active one
validation['plex'] = bool(self.get('plex.base_url')) and bool(self.get('plex.token'))
validation['jellyfin'] = bool(self.get('jellyfin.base_url')) and bool(self.get('jellyfin.api_key'))
@ -143,4 +331,4 @@ class ConfigManager:
return validation
config_manager = ConfigManager()
config_manager = ConfigManager()