""" db_manager.py — Configuration Storage Layer (Multi-Backend) Storage priority: 1. sqlite3 Python module (preferred, full dpworldapp compat) 2. sqlite3 CLI via subprocess (dpworldapp compat on minimal Python) 3. JSON file fallback (no dpworldapp compat, temp only) """ import os import sys import json import subprocess import shutil import threading import time from datetime import datetime, timezone try: import sqlite3 HAS_SQLITE = True except ImportError: HAS_SQLITE = False DB_PATH_DEFAULT = os.path.join(os.path.expanduser("~"), "db", "dynamic_data.db") # save_config retries transient SQLite "database is locked" errors: a competing # writer can briefly hold the lock even with PRAGMA busy_timeout set. Total # attempts, and the pause between them. _WRITE_MAX_ATTEMPTS = 3 _WRITE_RETRY_DELAY_S = 0.1 def _has_sqlite3_cli(): """Check if sqlite3 command-line tool is available.""" return shutil.which("sqlite3") is not None class DBManager: """Configuration storage manager with 3-tier backend.""" # Whitelist of allowed config keys (SQL injection defense for CLI backend) ALLOWED_KEYS = frozenset({'device_config', 'protocol_config', 'log_config', 'net_config', 'ap_config', 'uplink_config'}) # uplink: Telemetry Uplink (web-owned) def __init__(self, db_path=None, backend=None): self.db_path = db_path or os.environ.get("DB_PATH", DB_PATH_DEFAULT) self._lock = threading.Lock() if backend is not None: # Explicit backend override (used in tests and special deployments) self._backend = backend if backend == "json": self._json_dir = os.path.dirname(self.db_path) os.makedirs(self._json_dir, exist_ok=True) print(f"[DB] Backend override: {backend} ({self.db_path})") elif HAS_SQLITE: self._backend = "python" print(f"[DB] Using Python sqlite3: {self.db_path}") elif _has_sqlite3_cli(): self._backend = "cli" print(f"[DB] Using sqlite3 CLI: {self.db_path}") else: self._backend = "json" self._json_dir = os.path.dirname(self.db_path) os.makedirs(self._json_dir, exist_ok=True) print(f"[DB] Fallback to JSON files: {self._json_dir}") # ─── Python sqlite3 module ─────────────────────────────────── def _connect(self): conn = sqlite3.connect(self.db_path) conn.execute("PRAGMA busy_timeout = 5000") conn.execute("PRAGMA journal_mode = WAL") conn.row_factory = sqlite3.Row return conn def _validate_key(self, key): """Validate config key against whitelist to prevent SQL injection.""" if not isinstance(key, str): raise ValueError(f"Invalid config key type: {type(key)}") if key not in self.ALLOWED_KEYS: raise ValueError(f"Invalid config key: {key}") # ─── sqlite3 CLI (subprocess) ──────────────────────────────── def _cli_exec(self, sql): """Execute SQL via sqlite3 CLI and return stdout.""" os.makedirs(os.path.dirname(self.db_path), exist_ok=True) result = subprocess.run( ["sqlite3", self.db_path], input=sql, capture_output=True, text=True, timeout=10 ) if result.returncode != 0: raise RuntimeError(f"sqlite3 CLI error: {result.stderr.strip()}") return result.stdout.strip() def _cli_query(self, sql): """Execute query via CLI and return result string.""" return self._cli_exec(sql) # ─── JSON file helpers ─────────────────────────────────────── def _json_path(self, key): return os.path.join(self._json_dir, f"{key}.json") def _read_json_file(self, key): path = self._json_path(key) if not os.path.exists(path): return None try: with open(path, "r", encoding="utf-8") as f: return json.load(f) except (json.JSONDecodeError, IOError): return None def _write_json_file(self, key, data): path = self._json_path(key) tmp_path = path + ".tmp" json_str = json.dumps(data, ensure_ascii=False, indent=2) with open(tmp_path, "w", encoding="utf-8") as f: f.write(json_str) f.flush() os.fsync(f.fileno()) os.replace(tmp_path, path) # ─── Public API ────────────────────────────────────────────── def ensure_tables(self): if self._backend == "python": with self._lock: conn = self._connect() try: conn.execute(""" CREATE TABLE IF NOT EXISTS board_config ( key TEXT PRIMARY KEY NOT NULL, value TEXT NOT NULL ) """) conn.execute(""" CREATE TABLE IF NOT EXISTS event_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, data TEXT NOT NULL ) """) conn.commit() finally: conn.close() elif self._backend == "cli": with self._lock: self._cli_exec( "CREATE TABLE IF NOT EXISTS board_config " "(key TEXT PRIMARY KEY NOT NULL, value TEXT NOT NULL);\n" "CREATE TABLE IF NOT EXISTS event_history " "(id INTEGER PRIMARY KEY AUTOINCREMENT, data TEXT NOT NULL);" ) else: os.makedirs(self._json_dir, exist_ok=True) def get_config(self, key): self._validate_key(key) if self._backend == "python": with self._lock: conn = self._connect() try: cursor = conn.execute( "SELECT value FROM board_config WHERE key = ?", (key,) ) row = cursor.fetchone() if row is None: return None try: return json.loads(row["value"]) except json.JSONDecodeError: print(f"[DB] WARNING: malformed JSON in board_config key={key!r}; returning None", file=sys.stderr) return None finally: conn.close() elif self._backend == "cli": with self._lock: raw = self._cli_query( f"SELECT value FROM board_config WHERE key = '{key}';" ) if not raw: return None try: return json.loads(raw) except json.JSONDecodeError: print(f"[DB] WARNING: malformed JSON in board_config key={key!r}; returning None", file=sys.stderr) return None else: with self._lock: return self._read_json_file(key) def save_config(self, key, data): self._validate_key(key) json_str = json.dumps(data, ensure_ascii=False, indent=2) if self._backend == "python": last_error = None for attempt in range(_WRITE_MAX_ATTEMPTS): # #37 fix: acquire the lock only around the actual DB op; the retry # backoff sleep below runs OUTSIDE the lock so a contended write # does not stall every other DB thread for the backoff window. with self._lock: conn = self._connect() try: conn.execute( "INSERT OR REPLACE INTO board_config (key, value) VALUES (?, ?)", (key, json_str) ) conn.commit() return except sqlite3.OperationalError as e: # Retry only the transient lock; re-raise anything else. if "database is locked" not in str(e).lower(): raise last_error = e finally: conn.close() if attempt < _WRITE_MAX_ATTEMPTS - 1: time.sleep(_WRITE_RETRY_DELAY_S) # outside the lock raise last_error elif self._backend == "cli": # Escape single quotes in JSON for SQL escaped = json_str.replace("'", "''") with self._lock: self._cli_exec( f"INSERT OR REPLACE INTO board_config (key, value) " f"VALUES ('{key}', '{escaped}');" ) else: with self._lock: self._write_json_file(key, data) def update_config(self, key, mutator, default=None): """ Atomic read-modify-write of board_config entry. Args: key: board_config key (must be in ALLOWED_KEYS) mutator: callable(current_value) -> new_value. current_value is the JSON-deserialized current value (or `default` if absent). Must NOT mutate input — return a new dict. default: value to pass to mutator when key is missing (default: None) Returns: The new value (after mutator) — i.e., what was actually saved. If mutator returns None the key is deleted and None is returned. Atomic guarantees: - python backend: BEGIN IMMEDIATE + SELECT + UPDATE + COMMIT within one connection. Other writers block; lock retries with exponential backoff (same _WRITE_MAX_ATTEMPTS pattern). - cli backend: serialized by _lock (cli is sequential). - json backend: lock + read + mutator + atomic write (tempfile+rename). """ self._validate_key(key) if self._backend == "python": last_error = None for attempt in range(_WRITE_MAX_ATTEMPTS): # #37 fix: lock only around the atomic RMW; the retry backoff sleep # below runs OUTSIDE the lock so a contended write does not stall # every other DB thread for the backoff window. with self._lock: conn = self._connect() try: # BEGIN IMMEDIATE — acquires write lock immediately, # blocking other writers but allowing concurrent reads. conn.execute("BEGIN IMMEDIATE") cursor = conn.execute( "SELECT value FROM board_config WHERE key = ?", (key,) ) row = cursor.fetchone() if row is None: current = default else: try: current = json.loads(row["value"]) except json.JSONDecodeError: print(f"[DB] WARNING: malformed JSON in board_config key={key!r}; using default", file=sys.stderr) current = default new_value = mutator(current) if new_value is None: # mutator returned None → delete key conn.execute( "DELETE FROM board_config WHERE key = ?", (key,) ) else: json_str = json.dumps(new_value, ensure_ascii=False, indent=2) conn.execute( "INSERT OR REPLACE INTO board_config (key, value) VALUES (?, ?)", (key, json_str) ) conn.commit() return new_value except sqlite3.OperationalError as e: if "database is locked" not in str(e).lower(): raise last_error = e try: conn.rollback() except Exception: pass finally: conn.close() if attempt < _WRITE_MAX_ATTEMPTS - 1: time.sleep(_WRITE_RETRY_DELAY_S) # outside the lock raise last_error elif self._backend == "cli": with self._lock: raw = self._cli_query( f"SELECT value FROM board_config WHERE key = '{key}';" ) if not raw: current = default else: try: current = json.loads(raw) except json.JSONDecodeError: print(f"[DB] WARNING: malformed JSON in board_config key={key!r}; using default", file=sys.stderr) current = default new_value = mutator(current) if new_value is None: self._cli_exec(f"DELETE FROM board_config WHERE key = '{key}';") else: json_str = json.dumps(new_value, ensure_ascii=False, indent=2) escaped = json_str.replace("'", "''") self._cli_exec( f"INSERT OR REPLACE INTO board_config (key, value) " f"VALUES ('{key}', '{escaped}');" ) return new_value else: # json backend with self._lock: current = self._read_json_file(key) if current is None: current = default new_value = mutator(current) if new_value is None: path = self._json_path(key) if os.path.exists(path): os.remove(path) else: self._write_json_file(key, new_value) return new_value def config_exists(self, key): self._validate_key(key) if self._backend == "python": with self._lock: conn = self._connect() try: cursor = conn.execute( "SELECT 1 FROM board_config WHERE key = ? LIMIT 1", (key,) ) return cursor.fetchone() is not None finally: conn.close() elif self._backend == "cli": with self._lock: raw = self._cli_query( f"SELECT 1 FROM board_config WHERE key = '{key}' LIMIT 1;" ) return bool(raw) else: with self._lock: return os.path.exists(self._json_path(key)) # ─── Schema meta (migration flags) ────────────────────────── def _ensure_schema_meta_table(self) -> None: # _lock is acquired here independently; callers must NOT hold the # lock when calling this method (sequential acquisition only). with self._lock: conn = self._connect() try: conn.execute(""" CREATE TABLE IF NOT EXISTS schema_meta ( key TEXT PRIMARY KEY, value TEXT NOT NULL, applied_at TEXT NOT NULL ) """) conn.commit() finally: conn.close() def get_schema_meta(self, key: str): self._ensure_schema_meta_table() with self._lock: conn = self._connect() try: row = conn.execute( "SELECT value FROM schema_meta WHERE key = ?", (key,) ).fetchone() return row[0] if row else None finally: conn.close() def set_schema_meta(self, key: str, value: str) -> None: self._ensure_schema_meta_table() now = datetime.now(timezone.utc).isoformat() with self._lock: conn = self._connect() try: conn.execute(""" INSERT INTO schema_meta(key, value, applied_at) VALUES (?, ?, ?) ON CONFLICT(key) DO UPDATE SET value=excluded.value, applied_at=excluded.applied_at """, (key, value, now)) conn.commit() finally: conn.close() def get_raw_json(self, key): self._validate_key(key) if self._backend == "python": with self._lock: conn = self._connect() try: cursor = conn.execute( "SELECT value FROM board_config WHERE key = ?", (key,) ) row = cursor.fetchone() return row["value"] if row else None finally: conn.close() elif self._backend == "cli": with self._lock: raw = self._cli_query( f"SELECT value FROM board_config WHERE key = '{key}';" ) return raw if raw else None else: with self._lock: data = self._read_json_file(key) return json.dumps(data, ensure_ascii=False, indent=2) if data else None