liumangmang
62 lines
1.6 KiB
Python
62 lines
1.6 KiB
Python
import json
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
from .defaults import create_default_config, merge_config
|
|
|
|
|
|
class ConfigStore:
|
|
def __init__(self, db_path: Path):
|
|
self.db_path = Path(db_path)
|
|
self.db_path.parent.mkdir(parents=True, exist_ok=True)
|
|
self._initialize()
|
|
|
|
def _connect(self):
|
|
connection = sqlite3.connect(self.db_path)
|
|
connection.row_factory = sqlite3.Row
|
|
return connection
|
|
|
|
def _initialize(self):
|
|
with self._connect() as connection:
|
|
connection.execute(
|
|
'''
|
|
CREATE TABLE IF NOT EXISTS app_config (
|
|
id INTEGER PRIMARY KEY CHECK (id = 1),
|
|
config_json TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
'''
|
|
)
|
|
connection.commit()
|
|
|
|
def get_config(self):
|
|
with self._connect() as connection:
|
|
row = connection.execute(
|
|
'SELECT config_json FROM app_config WHERE id = 1'
|
|
).fetchone()
|
|
|
|
if not row:
|
|
config = create_default_config()
|
|
self.save_config(config)
|
|
return config
|
|
|
|
return merge_config(json.loads(row['config_json']))
|
|
|
|
def save_config(self, config):
|
|
normalized_config = merge_config(config)
|
|
|
|
with self._connect() as connection:
|
|
connection.execute(
|
|
'''
|
|
INSERT INTO app_config (id, config_json, updated_at)
|
|
VALUES (1, ?, CURRENT_TIMESTAMP)
|
|
ON CONFLICT(id) DO UPDATE SET
|
|
config_json = excluded.config_json,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
''',
|
|
(json.dumps(normalized_config),)
|
|
)
|
|
connection.commit()
|
|
|
|
return normalized_config
|