webapp/app/services/card_service.py

# app/services/card_service.py

from typing import Optional

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.card import Card
from app.schemas.card import CardCreate
from app.utils.search import apply_keyword_search_stmt
from app.utils.pagination import paginate
from app.core.biz_exception import BizLogicError


class CardService:

    # =========================
    # 创建 Card
    # =========================
    @staticmethod
    async def create(
        db: AsyncSession,
        obj_in: CardCreate,
    ) -> Card:
        db_obj = Card(**obj_in.dict())
        db.add(db_obj)
        await db.commit()
        await db.refresh(db_obj)
        return db_obj

    # =========================
    # 关键字分页查询
    # =========================
    @staticmethod
    async def list_by_keyword(
        db: AsyncSession,
        keyword: Optional[str] = None,
        page: int = 0,
        size: int = 10,
        culture: str = "english",
    ):
        stmt = select(Card).where(Card.culture == culture)

        stmt = apply_keyword_search_stmt(
            stmt=stmt,
            model=Card,
            keyword=keyword,
            fields=["title", "content", "label"],
        )

        return await paginate(db, stmt, page, size)