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

from datetime import datetime, timezone

from sqlalchemy.orm import Session

from packages.adapters.sqlalchemy_impl.models import PointsOrderModel
from packages.domain.points_order import PointsOrder


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

    def create(self, order: PointsOrder) -> PointsOrder:
        model = PointsOrderModel(
            id=order.id,
            user_id=order.user_id,
            order_type=order.order_type,
            product_code=order.product_code,
            amount_cents=order.amount_cents,
            original_amount_cents=order.original_amount_cents,
            discount=order.discount,
            points_amount=order.points_amount,
            status=order.status,
            payment_method=order.payment_method,
            payment_id=order.payment_id,
            paid_at=order.paid_at,
            created_at=order.created_at,
        )
        self.session.add(model)
        self.session.commit()
        return order

    def get(self, order_id: str) -> PointsOrder | None:
        model = (
            self.session.query(PointsOrderModel)
            .filter(PointsOrderModel.id == order_id)
            .first()
        )
        if model is None:
            return None
        return self._to_domain(model)

    def update_status(
        self,
        order_id: str,
        status: str,
        *,
        payment_id: str | None = None,
        paid_at: datetime | None = None,
    ) -> PointsOrder | None:
        model = (
            self.session.query(PointsOrderModel)
            .filter(PointsOrderModel.id == order_id)
            .first()
        )
        if model is None:
            return None
        model.status = status
        if payment_id is not None:
            model.payment_id = payment_id
        if paid_at is not None:
            model.paid_at = paid_at
        self.session.add(model)
        self.session.commit()
        return self._to_domain(model)

    def list_by_user(
        self,
        user_id: str,
        *,
        order_type: str | None = None,
        status: str | None = None,
        page: int = 1,
        page_size: int = 20,
    ) -> tuple[list[PointsOrder], int]:
        query = self.session.query(PointsOrderModel).filter(
            PointsOrderModel.user_id == user_id
        )
        if order_type:
            query = query.filter(PointsOrderModel.order_type == order_type)
        if status:
            query = query.filter(PointsOrderModel.status == status)

        total = query.count()
        models = (
            query.order_by(PointsOrderModel.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: PointsOrderModel) -> PointsOrder:
        return PointsOrder(
            id=model.id,
            user_id=model.user_id,
            order_type=model.order_type,
            product_code=model.product_code,
            amount_cents=model.amount_cents,
            original_amount_cents=model.original_amount_cents,
            discount=model.discount,
            points_amount=model.points_amount,
            status=model.status,
            payment_method=model.payment_method,
            payment_id=model.payment_id,
            paid_at=model.paid_at,
            created_at=model.created_at,
        )