Cheng Zhou
59 lines
1.9 KiB
Python
59 lines
1.9 KiB
Python
import uuid
|
|
from typing import Sequence
|
|
|
|
from sqlalchemy import select, update as sa_update
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.faq_category import FaqCategory
|
|
from app.schemas.faq import CategoryCreate, CategoryUpdate
|
|
|
|
|
|
async def create_category(db: AsyncSession, category_in: CategoryCreate) -> FaqCategory:
|
|
category = FaqCategory(
|
|
study_id=category_in.study_id,
|
|
name=category_in.name,
|
|
description=category_in.description,
|
|
sort_order=category_in.sort_order,
|
|
is_active=category_in.is_active,
|
|
)
|
|
db.add(category)
|
|
await db.commit()
|
|
await db.refresh(category)
|
|
return category
|
|
|
|
|
|
async def list_categories(
|
|
db: AsyncSession,
|
|
study_id: uuid.UUID | None,
|
|
include_global: bool = True,
|
|
is_active: bool | None = True,
|
|
) -> Sequence[FaqCategory]:
|
|
stmt = select(FaqCategory)
|
|
if include_global:
|
|
stmt = stmt.where((FaqCategory.study_id == study_id) | (FaqCategory.study_id.is_(None)))
|
|
else:
|
|
stmt = stmt.where(FaqCategory.study_id == study_id)
|
|
if is_active is not None:
|
|
stmt = stmt.where(FaqCategory.is_active == is_active)
|
|
stmt = stmt.order_by(FaqCategory.sort_order, FaqCategory.name)
|
|
result = await db.execute(stmt)
|
|
return result.scalars().all()
|
|
|
|
|
|
async def get_category(db: AsyncSession, category_id: uuid.UUID) -> FaqCategory | None:
|
|
result = await db.execute(select(FaqCategory).where(FaqCategory.id == category_id))
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def update_category(db: AsyncSession, category: FaqCategory, category_in: CategoryUpdate) -> FaqCategory:
|
|
update_data = category_in.model_dump(exclude_unset=True)
|
|
if update_data:
|
|
await db.execute(
|
|
sa_update(FaqCategory)
|
|
.where(FaqCategory.id == category.id)
|
|
.values(**update_data)
|
|
)
|
|
await db.commit()
|
|
await db.refresh(category)
|
|
return category
|