webapp/app/services/product_service.py

# app/services/product_service.py
from sqlalchemy.orm import Session
from typing import Optional, List
from app.utils.search import apply_keyword_search
from app.utils.pagination import paginate
from app.core.biz_exception import NotFoundError, PermissionDeniedError, BizLogicError
from app.models.product import VasProduct
from app.schemas.product import VasProductCreate, VasProductUpdate, VasProductOut

class ProductService:

    def create(db: Session, data: VasProductCreate):
        rec = VasProduct(**data.dict())
        db.add(rec)
        db.commit()
        db.refresh(rec)
        return rec

    def get(db: Session, id:int):
        obj = db.query(VasProduct).filter_by(id=id).first()
        if not obj:
            raise NotFoundError('Product not exist')
        return obj

    def update(db: Session, id:int, data: VasProductUpdate):
        rec = db.query(VasProduct).filter_by(id=id).first()
        if not rec:
            raise NotFoundError('Product not exist')
        for k,v in data.dict(exclude_unset=True).items():
            setattr(rec,k,v)
        db.commit()
        db.refresh(rec)
        return rec
    
    def list_product(db: Session, country: str = None, visa_type: str = None, page: int=0, size: int=10, keyword: str=None):    
        query = db.query(VasProduct)
        
        if country:
            query = query.filter(VasProduct.country == country)
            
        if visa_type:
            query = query.filter(VasProduct.visa_type == visa_type)
        
        query = apply_keyword_search(
            query=query,
            model=VasProduct,
            keyword=keyword,
            fields=["title", "provider", "description"]
        )
        return paginate(query, page, size)