webapp/app/services/payment_qr_service.py

# app/services/payment_qr_service.py
from sqlalchemy.orm import Session
from app.core.biz_exception import NotFoundError, PermissionDeniedError, BizLogicError
from app.models.payment_provider import VasPaymentProvider
from app.models.payment_qr import VasPaymentQR
from app.schemas.payment_qr import VasPaymentQrCreate, VasPaymentQrSetEnableIn

class PaymentQrService:

    def create(db: Session, data: VasPaymentQrCreate):
        rec = VasPaymentQR(**data.dict())
        db.add(rec)
        db.commit()
        db.refresh(rec)
        return rec
    
    def get_by_id(db: Session, id: int):
        obj = db.query(VasPaymentQR).filter(VasPaymentQR.id == id).first()
        if not obj:
            raise NotFoundError("QR not exist")
        return obj
    
    def set_enable(db: Session, id: int, payload: VasPaymentQrSetEnableIn):
        obj = db.query(VasPaymentQR).filter(VasPaymentQR.id == id).first()
        if not obj:
            raise NotFoundError("QR not exist")
        obj.is_active = payload.is_active
        db.commit()
        db.refresh(obj)
        return obj
    
    def delete(db: Session, id: int):
        obj = db.query(VasPaymentQR).filter(VasPaymentQR.id == id).first()
        if not obj:
            raise NotFoundError("QR not exist")
        db.delete(obj)
        db.commit()
    
    def get_by_devid(db: Session, devid: str):
        return db.query(VasPaymentQR).filter(VasPaymentQR.devid == devid).all()

    def get_by_provider(db: Session, provider: str):
        return db.query(VasPaymentQR).filter(VasPaymentQR.provider == provider).all()
    
    def list_by_provider(db: Session, provider_id: int):
        obj = db.query(VasPaymentProvider).filter(VasPaymentProvider.id==provider_id).first()
        if not obj:
            raise NotFoundError("Provider not exist")
        
        return db.query(VasPaymentQR).filter(VasPaymentQR.provider==obj.name).all()