"""运营配置读写。 get_value:业务读配置,DB 没有则 fallback 到 CONFIG_DEFS 默认(= 原 rewards 常量,空表行为不变)。 不缓存:配置读频率低(每次福利操作读一次,key 主键查极快),admin 改了立即生效、跨进程一致。 """ from __future__ import annotations from typing import Any from sqlalchemy import select from sqlalchemy.orm import Session from app.core.config_schema import CONFIG_DEFS from app.models.app_config import AppConfig def get_value(db: Session, key: str) -> Any: """读配置值。DB 有用 DB,没有 fallback 到 CONFIG_DEFS 默认。未知 key 抛 KeyError。""" row = db.get(AppConfig, key) if row is not None: return row.value return CONFIG_DEFS[key]["default"] def set_value( db: Session, key: str, value: Any, *, admin_id: int, commit: bool = True ) -> AppConfig: """写配置(admin 用)。未知 key 抛 KeyError。commit=False 让调用方与审计同事务。""" if key not in CONFIG_DEFS: raise KeyError(f"unknown config key: {key}") row = db.get(AppConfig, key) if row is None: row = AppConfig(key=key, value=value, updated_by_admin_id=admin_id) db.add(row) else: row.value = value row.updated_by_admin_id = admin_id if commit: db.commit() db.refresh(row) else: db.flush() return row def list_all(db: Session) -> list[dict]: """所有可配项:default + 当前值(DB 有则用 DB) + 元信息。给 admin 配置页渲染。""" overrides = {r.key: r for r in db.execute(select(AppConfig)).scalars().all()} out: list[dict] = [] for key, defn in CONFIG_DEFS.items(): row = overrides.get(key) out.append( { "key": key, "label": defn["label"], "group": defn["group"], "type": defn["type"], "help": defn.get("help"), "default": defn["default"], "value": row.value if row is not None else defn["default"], "overridden": row is not None, "updated_at": row.updated_at.isoformat() if row is not None else None, } ) return out