personal-consumption-service/app/repositories/invoice_repo.py

# @description: 
# @author: licanglong
# @date: 2025/12/22 20:16

from sqlalchemy import select
from sqlalchemy.orm import Session

from app.models.dto import InvoicePurchaseDetailDTO, InvoicePurchaseDetailList
from app.models.entity import InvoicePurchaseDetail


class InvoiceRepository:
    def __init__(self, session: Session):
        self.session = session

    async def list_by_taxid(self, tax_id: str) -> InvoicePurchaseDetailList:
        stmt = select(InvoicePurchaseDetail).where(InvoicePurchaseDetail.tax_id == tax_id).limit(10)
        stmt_result = self.session.scalars(stmt).all()
        return InvoicePurchaseDetailList(
            invoice_purchase_details=[
                InvoicePurchaseDetailDTO.model_validate(record)
                for record in stmt_result
            ]
        )

    async def list_by_params(self, tax_id: str = None, hwmc: str = None, page_number: int = 1,
                             page_size: int = 10) -> InvoicePurchaseDetailList:
        conditions = []
        if tax_id:
            conditions.append(InvoicePurchaseDetail.tax_id == tax_id)

        if hwmc:
            conditions.append(InvoicePurchaseDetail.hwmc.like(f"%{hwmc}%"))
        stmt = select(InvoicePurchaseDetail)

        if conditions:
            stmt = stmt.where(*conditions)
        stmt = stmt.offset((page_number - 1) * page_size).limit(page_size)
        stmt_result = self.session.scalars(stmt).all()
        return InvoicePurchaseDetailList(
            invoice_purchase_details=[
                InvoicePurchaseDetailDTO.model_validate(record)
                for record in stmt_result
            ]
        )