payment-platform/backend/app/plugin/module_payment/points/crud.py

from app.api.v1.module_system.auth.schema import AuthSchema
from app.core.base_crud import CRUDBase
from app.core.exceptions import CustomException

from .model import PointsModel, PointsRecordModel
from .schema import PointsCreateSchema


class PointsCRUD(CRUDBase[PointsModel, PointsCreateSchema, PointsCreateSchema]):
    """积分 CRUD 操作"""

    def __init__(self, auth: AuthSchema) -> None:
        self.auth = auth
        super().__init__(model=PointsModel, auth=auth)

    async def get_by_tenant_id(self, tenant_id: int) -> PointsModel | None:
        """根据租户ID查询积分"""
        return await self.get(tenant_id=tenant_id)

    async def create_or_update(self, tenant_id: int, points: float) -> PointsModel:
        """创建或更新积分"""
        points_obj = await self.get_by_tenant_id(tenant_id)
        if points_obj:
            points_obj.points = points
            await self.auth.db.flush()
            await self.auth.db.refresh(points_obj)
            return points_obj
        else:
            create_data = {
                "tenant_id": tenant_id,
                "points": points
            }
            return await self.create(create_data)


class PointsRecordCRUD(CRUDBase[PointsRecordModel, None, None]):
    """积分记录 CRUD 操作"""

    def __init__(self, auth: AuthSchema) -> None:
        self.auth = auth
        super().__init__(model=PointsRecordModel, auth=auth)