webapp/app/services/product_routing_service.py

# app/services/product_routing_service.py

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

from app.core.biz_exception import NotFoundError
from app.models.product_routing import VasProductRouting
from app.schemas.product_routing import VasProductRoutingCreate


class ProductRoutingService:

    @staticmethod
    async def create(
        db: AsyncSession,
        data: VasProductRoutingCreate,
    ) -> VasProductRouting:
        rec = VasProductRouting(**data.dict())
        db.add(rec)
        await db.commit()
        await db.refresh(rec)
        return rec

    @staticmethod
    async def list_by_product(
        db: AsyncSession,
        product_id: int,
    ):
        stmt = select(VasProductRouting).where(
            VasProductRouting.product_id == product_id
        )
        result = await db.execute(stmt)
        return result.scalars().all()

    @staticmethod
    async def delete(
        db: AsyncSession,
        id: int,
    ):
        stmt = select(VasProductRouting).where(
            VasProductRouting.id == id
        )
        obj = (await db.execute(stmt)).scalar_one_or_none()
        if not obj:
            raise NotFoundError("Product routing not exist")

        await db.delete(obj)
        await db.commit()