2025-12-10 12:02:17 +08:00

39 lines
1.2 KiB
Python

from typing import List, Optional
from fastapi import HTTPException, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from backend.modules.settings.models import Setting
async def list_settings(session: AsyncSession) -> List[Setting]:
return (await session.scalars(select(Setting))).all()
async def create_setting(session: AsyncSession, data: dict) -> Setting:
setting = Setting(**data)
session.add(setting)
await session.commit()
await session.refresh(setting)
return setting
async def update_setting(session: AsyncSession, setting_id: int, data: dict) -> Setting:
setting = await session.get(Setting, setting_id)
if not setting:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Setting not found")
for field, value in data.items():
setattr(setting, field, value)
await session.commit()
await session.refresh(setting)
return setting
async def delete_setting(session: AsyncSession, setting_id: int) -> None:
setting = await session.get(Setting, setting_id)
if not setting:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Setting not found")
await session.delete(setting)
await session.commit()