97 lines
3.2 KiB
Python
97 lines
3.2 KiB
Python
|
"""非敏感应用设置的 SQLite Repository。"""
|
||
|
|
|
||
|
|
import json
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Any, Optional, Union
|
||
|
|
|
||
|
|
from .db import initialize_database, open_database
|
||
|
|
from .task_models import AppSettingRecord
|
||
|
|
|
||
|
|
|
||
|
|
PathValue = Union[str, Path]
|
||
|
|
|
||
|
|
|
||
|
|
def _utc_now_iso() -> str:
|
||
|
|
return datetime.now(timezone.utc).isoformat(timespec="seconds").replace(
|
||
|
|
"+00:00", "Z"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
class SettingsRepository:
|
||
|
|
"""用 JSON 保存非敏感设置；凭据不得传入本类。"""
|
||
|
|
|
||
|
|
def __init__(self, db_path: Optional[PathValue] = None):
|
||
|
|
self._db_path = initialize_database(db_path)
|
||
|
|
|
||
|
|
def get(self, setting_key: str, default: Any = None) -> Any:
|
||
|
|
"""读取设置值；键不存在时返回 default。"""
|
||
|
|
|
||
|
|
record = self.get_record(setting_key)
|
||
|
|
return default if record is None else record.value
|
||
|
|
|
||
|
|
def get_record(self, setting_key: str) -> Optional[AppSettingRecord]:
|
||
|
|
"""读取包含更新时间的完整设置记录。"""
|
||
|
|
|
||
|
|
key = self._validate_key(setting_key)
|
||
|
|
connection = open_database(self._db_path)
|
||
|
|
try:
|
||
|
|
row = connection.execute(
|
||
|
|
"SELECT setting_key, value_json, updated_at"
|
||
|
|
" FROM app_settings WHERE setting_key = ?",
|
||
|
|
(key,),
|
||
|
|
).fetchone()
|
||
|
|
finally:
|
||
|
|
connection.close()
|
||
|
|
if row is None:
|
||
|
|
return None
|
||
|
|
return AppSettingRecord(
|
||
|
|
setting_key=row["setting_key"],
|
||
|
|
value=json.loads(row["value_json"]),
|
||
|
|
updated_at=row["updated_at"],
|
||
|
|
)
|
||
|
|
|
||
|
|
def set(
|
||
|
|
self, setting_key: str, value: Any, updated_at: Optional[str] = None
|
||
|
|
) -> AppSettingRecord:
|
||
|
|
"""新增或更新一个非敏感设置并返回保存后的记录。"""
|
||
|
|
|
||
|
|
key = self._validate_key(setting_key)
|
||
|
|
now = updated_at or _utc_now_iso()
|
||
|
|
value_json = json.dumps(value, ensure_ascii=False)
|
||
|
|
connection = open_database(self._db_path)
|
||
|
|
try:
|
||
|
|
with connection:
|
||
|
|
connection.execute(
|
||
|
|
"INSERT INTO app_settings (setting_key, value_json, updated_at)"
|
||
|
|
" VALUES (?, ?, ?)"
|
||
|
|
" ON CONFLICT(setting_key) DO UPDATE SET"
|
||
|
|
" value_json = excluded.value_json,"
|
||
|
|
" updated_at = excluded.updated_at",
|
||
|
|
(key, value_json, now),
|
||
|
|
)
|
||
|
|
finally:
|
||
|
|
connection.close()
|
||
|
|
return AppSettingRecord(key, value, now)
|
||
|
|
|
||
|
|
def delete(self, setting_key: str) -> bool:
|
||
|
|
"""删除设置；确实删除了一条记录时返回 True。"""
|
||
|
|
|
||
|
|
key = self._validate_key(setting_key)
|
||
|
|
connection = open_database(self._db_path)
|
||
|
|
try:
|
||
|
|
with connection:
|
||
|
|
cursor = connection.execute(
|
||
|
|
"DELETE FROM app_settings WHERE setting_key = ?", (key,)
|
||
|
|
)
|
||
|
|
return cursor.rowcount > 0
|
||
|
|
finally:
|
||
|
|
connection.close()
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def _validate_key(setting_key: str) -> str:
|
||
|
|
key = setting_key.strip()
|
||
|
|
if not key:
|
||
|
|
raise ValueError("setting_key 不能为空")
|
||
|
|
return key
|