xiaoxia-saas/packages/adapters/sqlalchemy_impl/points_transaction_repository.py

from sqlalchemy.orm import Session

from packages.adapters.sqlalchemy_impl.models import PointsTransactionModel
from packages.domain.points_transaction import PointsTransaction


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

    def create(self, transaction: PointsTransaction) -> PointsTransaction:
        model = PointsTransactionModel(
            id=transaction.id,
            user_id=transaction.user_id,
            account_id=transaction.account_id,
            type=transaction.type,
            source=transaction.source,
            amount=transaction.amount,
            balance_after=transaction.balance_after,
            description=transaction.description,
            ref_id=transaction.ref_id,
            created_at=transaction.created_at,
        )
        self.session.add(model)
        self.session.commit()
        return transaction

    def list_by_user(
        self,
        user_id: str,
        *,
        type: str | None = None,
        source: str | None = None,
        page: int = 1,
        page_size: int = 20,
    ) -> tuple[list[PointsTransaction], int]:
        query = self.session.query(PointsTransactionModel).filter(PointsTransactionModel.user_id == user_id)
        if type:
            query = query.filter(PointsTransactionModel.type == type)
        if source:
            query = query.filter(PointsTransactionModel.source == source)

        total = query.count()
        models = (
            query.order_by(PointsTransactionModel.created_at.desc())
            .offset((page - 1) * page_size)
            .limit(page_size)
            .all()
        )
        return [self._to_domain(m) for m in models], total

    @staticmethod
    def _to_domain(model: PointsTransactionModel) -> PointsTransaction:
        return PointsTransaction(
            id=model.id,
            user_id=model.user_id,
            account_id=model.account_id,
            type=model.type,
            source=model.source,
            amount=model.amount,
            balance_after=model.balance_after,
            description=model.description or "",
            ref_id=model.ref_id or "",
            created_at=model.created_at,
        )