105 lines
2.9 KiB
Python
105 lines
2.9 KiB
Python
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.desktopConfig import DesktopConfig
|
|
from app.schemas.admin_desktop_config import DesktopConfigCreate, DesktopConfigUpdate
|
|
|
|
DEFAULT_PAGE_SIZE = 20
|
|
MAX_PAGE_SIZE = 200
|
|
|
|
|
|
class AdminDesktopConfigError(Exception):
|
|
def __init__(self, message: str) -> None:
|
|
self.message = message
|
|
super().__init__(message)
|
|
|
|
|
|
async def list_configs(
|
|
db: AsyncSession,
|
|
*,
|
|
page: int = 1,
|
|
page_size: int = DEFAULT_PAGE_SIZE,
|
|
keyword: str | None = None,
|
|
) -> tuple[list[DesktopConfig], int]:
|
|
page_size = min(max(page_size, 1), MAX_PAGE_SIZE)
|
|
page = max(page, 1)
|
|
offset = (page - 1) * page_size
|
|
|
|
filters = []
|
|
if keyword:
|
|
like = f"%{keyword.strip()}%"
|
|
filters.append(DesktopConfig.name.ilike(like))
|
|
|
|
count_stmt = select(func.count()).select_from(DesktopConfig)
|
|
list_stmt = select(DesktopConfig)
|
|
for clause in filters:
|
|
count_stmt = count_stmt.where(clause)
|
|
list_stmt = list_stmt.where(clause)
|
|
|
|
total = await db.scalar(count_stmt) or 0
|
|
result = await db.scalars(
|
|
list_stmt.order_by(DesktopConfig.name.asc()).offset(offset).limit(page_size)
|
|
)
|
|
return list(result.all()), total
|
|
|
|
|
|
async def create_config(db: AsyncSession, body: DesktopConfigCreate) -> DesktopConfig:
|
|
existing = await db.scalar(
|
|
select(DesktopConfig).where(
|
|
DesktopConfig.name == body.name,
|
|
DesktopConfig.oem_id == body.oem_id,
|
|
)
|
|
)
|
|
if existing is not None:
|
|
raise AdminDesktopConfigError("该 OEM 下配置键名已存在")
|
|
|
|
row = DesktopConfig(
|
|
name=body.name,
|
|
value=body.value,
|
|
mark=body.mark,
|
|
oem_id=body.oem_id,
|
|
)
|
|
db.add(row)
|
|
await db.flush()
|
|
await db.refresh(row)
|
|
return row
|
|
|
|
|
|
async def update_config(
|
|
db: AsyncSession,
|
|
config_id: int,
|
|
body: DesktopConfigUpdate,
|
|
) -> DesktopConfig:
|
|
row = await db.get(DesktopConfig, config_id)
|
|
if row is None:
|
|
raise AdminDesktopConfigError("配置不存在")
|
|
|
|
if body.name is not None and body.name != row.name:
|
|
taken = await db.scalar(
|
|
select(DesktopConfig).where(
|
|
DesktopConfig.name == body.name,
|
|
DesktopConfig.oem_id == row.oem_id,
|
|
DesktopConfig.id != config_id,
|
|
)
|
|
)
|
|
if taken is not None:
|
|
raise AdminDesktopConfigError("该 OEM 下配置键名已存在")
|
|
row.name = body.name
|
|
|
|
if body.value is not None:
|
|
row.value = body.value
|
|
|
|
if body.mark is not None:
|
|
row.mark = body.mark
|
|
|
|
await db.flush()
|
|
await db.refresh(row)
|
|
return row
|
|
|
|
|
|
async def delete_config(db: AsyncSession, config_id: int) -> None:
|
|
row = await db.get(DesktopConfig, config_id)
|
|
if row is None:
|
|
raise AdminDesktopConfigError("配置不存在")
|
|
await db.delete(row)
|