35 lines
1.1 KiB
Python
35 lines
1.1 KiB
Python
from collections.abc import Iterable
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.system_setting import SystemSetting
|
|
|
|
|
|
class SystemSettingsRepository:
|
|
@staticmethod
|
|
def get(db: Session, key: str) -> SystemSetting | None:
|
|
return db.scalar(select(SystemSetting).where(SystemSetting.key == key))
|
|
|
|
@staticmethod
|
|
def get_many(db: Session, keys: Iterable[str]) -> dict[str, SystemSetting]:
|
|
key_list = list(keys)
|
|
if not key_list:
|
|
return {}
|
|
|
|
settings = db.scalars(select(SystemSetting).where(SystemSetting.key.in_(key_list))).all()
|
|
return {setting.key: setting for setting in settings}
|
|
|
|
@staticmethod
|
|
def upsert(db: Session, *, key: str, value: str, is_secret: bool = False) -> SystemSetting:
|
|
setting = SystemSettingsRepository.get(db, key)
|
|
if setting is None:
|
|
setting = SystemSetting(key=key, value=value, is_secret=is_secret)
|
|
db.add(setting)
|
|
db.flush()
|
|
return setting
|
|
|
|
setting.value = value
|
|
setting.is_secret = is_secret
|
|
db.flush()
|
|
return setting
|