kodi
52 lines
1.5 KiB
Python
52 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
|
|
|
|
class SettingsRepository:
|
|
def __init__(self, db_path: str):
|
|
self._db_path = db_path
|
|
self._ensure_schema()
|
|
|
|
def get_settings(self) -> dict[str, str]:
|
|
with self._connection() as conn:
|
|
rows = conn.execute("SELECT key, value FROM settings").fetchall()
|
|
return {row["key"]: row["value"] for row in rows}
|
|
|
|
def set_setting(self, key: str, value: str) -> None:
|
|
with self._connection() as conn:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO settings (key, value)
|
|
VALUES (?, ?)
|
|
ON CONFLICT(key) DO UPDATE SET value = excluded.value
|
|
""",
|
|
(key, value),
|
|
)
|
|
|
|
def _ensure_schema(self) -> None:
|
|
db_path = Path(self._db_path)
|
|
if db_path.parent and str(db_path.parent) not in {"", "."}:
|
|
db_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with self._connection() as conn:
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS settings (
|
|
key TEXT PRIMARY KEY,
|
|
value TEXT NOT NULL
|
|
)
|
|
"""
|
|
)
|
|
|
|
@contextmanager
|
|
def _connection(self):
|
|
conn = sqlite3.connect(self._db_path)
|
|
conn.row_factory = sqlite3.Row
|
|
try:
|
|
yield conn
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|