# 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)