from typing import List, Optional from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select from pydantic import BaseModel from app.database import get_db from app.core.deps import require_role from app.models.user import User from app.models.setting import Setting router = APIRouter() class SettingResponse(BaseModel): key: str value: Optional[str] = None description: Optional[str] = None class SettingUpdate(BaseModel): value: str class BulkSettingUpdate(BaseModel): settings: List[SettingUpdate] @router.get("/") async def list_settings( db: AsyncSession = Depends(get_db), current_user: User = Depends(require_role(["admin"])) ): result = await db.execute(select(Setting)) settings = list(result.scalars().all()) return [ { "key": s.key, "value": s.value, "description": s.description } for s in settings ] @router.get("/{key}", response_model=SettingResponse) async def get_setting( key: str, db: AsyncSession = Depends(get_db), current_user: User = Depends(require_role(["admin"])) ): result = await db.execute(select(Setting).where(Setting.key == key)) setting = result.scalar_one_or_none() if not setting: raise HTTPException(status_code=404, detail="设置不存在") return SettingResponse( key=setting.key, value=setting.value, description=setting.description ) @router.put("/{key}", response_model=SettingResponse) async def update_setting( key: str, data: SettingUpdate, db: AsyncSession = Depends(get_db), current_user: User = Depends(require_role(["admin"])) ): result = await db.execute(select(Setting).where(Setting.key == key)) setting = result.scalar_one_or_none() if not setting: setting = Setting(key=key, value=data.value) db.add(setting) else: setting.value = data.value await db.commit() return SettingResponse( key=setting.key, value=setting.value, description=setting.description )