"""System configuration KV store operations.""" import time from typing import Optional, Any from storage.db import get_connection def get_config(key: str) -> Optional[str]: """Get a single config value.""" with get_connection() as conn: row = conn.execute( "SELECT value FROM system_config WHERE key = ?", (key,) ).fetchone() return row["value"] if row else None def set_config(key: str, value: str, description: str = "") -> None: """Set or update a config value.""" now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) with get_connection() as conn: conn.execute( """INSERT INTO system_config (key, value, description, updated_at) VALUES (?, ?, ?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value, description = excluded.description, updated_at = excluded.updated_at""", (key, value, description, now), ) conn.commit() def delete_config(key: str) -> bool: """Delete a config value.""" with get_connection() as conn: cursor = conn.execute( "DELETE FROM system_config WHERE key = ?", (key,) ) conn.commit() return cursor.rowcount > 0 def list_configs() -> list[dict]: """List all system config entries.""" with get_connection() as conn: rows = conn.execute("SELECT * FROM system_config ORDER BY key").fetchall() return [dict(row) for row in rows] def get_all_configs_as_dict() -> dict[str, str]: """Get all configs as a simple dict.""" with get_connection() as conn: rows = conn.execute("SELECT key, value FROM system_config").fetchall() return {row["key"]: row["value"] for row in rows}