webapp/app/services/payment_provider_service.py

# app/services/payment_provider.py

from typing import List, Optional

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

from app.core.biz_exception import NotFoundError, BizLogicError
from app.models.payment_provider import VasPaymentProvider
from app.schemas.payment_provider import (
    VasPaymentProviderCreate,
    VasPaymentProviderUpdate,
)


class PaymentProviderService:
    # --------------------------------------------------
    # 创建支付提供商（防重复）
    # --------------------------------------------------
    @staticmethod
    async def create(
        db: AsyncSession,
        data: VasPaymentProviderCreate,
    ) -> VasPaymentProvider:

        stmt = select(VasPaymentProvider).where(
            VasPaymentProvider.name == data.name,
            VasPaymentProvider.channel == data.channel,
            VasPaymentProvider.currency == data.currency,
        )
        exists = (await db.execute(stmt)).scalar_one_or_none()

        if exists:
            raise BizLogicError("Payment provider already exists")

        provider = VasPaymentProvider(
            name=data.name,
            channel=data.channel,
            currency=data.currency,
            icon=data.icon,
            enabled=data.enabled if data.enabled is not None else 1,
            config=data.config,
        )

        db.add(provider)
        await db.commit()
        await db.refresh(provider)
        return provider

    # --------------------------------------------------
    # 更新支付提供商（禁止修改 name/channel/currency）
    # --------------------------------------------------
    @staticmethod
    async def update(
        db: AsyncSession,
        provider_id: int,
        data: VasPaymentProviderUpdate,
    ) -> VasPaymentProvider:

        stmt = select(VasPaymentProvider).where(
            VasPaymentProvider.id == provider_id
        )
        provider = (await db.execute(stmt)).scalar_one_or_none()

        if not provider:
            raise NotFoundError("Payment provider not found")

        update_data = data.dict(exclude_unset=True)

        # 🚫 禁止修改三元组
        for forbidden in ("name", "channel", "currency"):
            update_data.pop(forbidden, None)

        for key, value in update_data.items():
            setattr(provider, key, value)

        await db.commit()
        await db.refresh(provider)
        return provider

    # --------------------------------------------------
    # 删除支付提供商
    # --------------------------------------------------
    @staticmethod
    async def delete(
        db: AsyncSession,
        provider_id: int,
    ) -> bool:

        stmt = select(VasPaymentProvider).where(
            VasPaymentProvider.id == provider_id
        )
        provider = (await db.execute(stmt)).scalar_one_or_none()

        if not provider:
            raise NotFoundError("Provider not exist")

        await db.delete(provider)
        await db.commit()
        return True

    # --------------------------------------------------
    # 所有支付提供商
    # --------------------------------------------------
    @staticmethod
    async def list_all(
        db: AsyncSession,
    ) -> List[VasPaymentProvider]:

        result = await db.execute(select(VasPaymentProvider))
        return result.scalars().all()

    # --------------------------------------------------
    # 可用的支付提供商（可按币种）
    # --------------------------------------------------
    @staticmethod
    async def list_enabled(
        db: AsyncSession,
        currency: Optional[str] = None,
    ) -> List[VasPaymentProvider]:

        stmt = select(VasPaymentProvider).where(
            VasPaymentProvider.enabled == 1
        )

        if currency:
            stmt = stmt.where(VasPaymentProvider.currency == currency)

        result = await db.execute(stmt)
        return result.scalars().all()

    # --------------------------------------------------
    # 根据 name 获取（只返回 enabled）
    # --------------------------------------------------
    @staticmethod
    async def get_by_name(
        db: AsyncSession,
        name: str,
    ) -> Optional[VasPaymentProvider]:

        stmt = select(VasPaymentProvider).where(
            VasPaymentProvider.enabled == 1,
            VasPaymentProvider.name == name,
        )

        return (await db.execute(stmt)).scalar_one_or_none()