rbox/mywebdav/billing/stripe_client.py

import stripe
from decimal import Decimal
from typing import Optional, Dict, Any
from ..settings import settings

class StripeClient:
    @staticmethod
    def _ensure_api_key():
        if not stripe.api_key:
            if settings.STRIPE_SECRET_KEY:
                stripe.api_key = settings.STRIPE_SECRET_KEY
            else:
                raise ValueError("Stripe API key not configured")
    @staticmethod
    async def create_customer(email: str, name: str, metadata: Dict = None) -> str:
        StripeClient._ensure_api_key()
        customer = stripe.Customer.create(
            email=email,
            name=name,
            metadata=metadata or {}
        )
        return customer.id

    @staticmethod
    async def create_payment_intent(
        amount: int,
        currency: str = "usd",
        customer_id: str = None,
        metadata: Dict = None
    ) -> stripe.PaymentIntent:
        StripeClient._ensure_api_key()
        return stripe.PaymentIntent.create(
            amount=amount,
            currency=currency,
            customer=customer_id,
            metadata=metadata or {},
            automatic_payment_methods={"enabled": True}
        )

    @staticmethod
    async def create_invoice(
        customer_id: str,
        description: str,
        line_items: list,
        metadata: Dict = None
    ) -> stripe.Invoice:
        StripeClient._ensure_api_key()
        for item in line_items:
            stripe.InvoiceItem.create(
                customer=customer_id,
                amount=int(item['amount'] * 100),
                currency=item.get('currency', 'usd'),
                description=item['description'],
                metadata=item.get('metadata', {})
            )

        invoice = stripe.Invoice.create(
            customer=customer_id,
            description=description,
            auto_advance=True,
            collection_method='charge_automatically',
            metadata=metadata or {}
        )

        return invoice

    @staticmethod
    async def finalize_invoice(invoice_id: str) -> stripe.Invoice:
        StripeClient._ensure_api_key()
        return stripe.Invoice.finalize_invoice(invoice_id)

    @staticmethod
    async def pay_invoice(invoice_id: str) -> stripe.Invoice:
        StripeClient._ensure_api_key()
        return stripe.Invoice.pay(invoice_id)

    @staticmethod
    async def attach_payment_method(
        payment_method_id: str,
        customer_id: str
    ) -> stripe.PaymentMethod:
        StripeClient._ensure_api_key()
        payment_method = stripe.PaymentMethod.attach(
            payment_method_id,
            customer=customer_id
        )

        stripe.Customer.modify(
            customer_id,
            invoice_settings={'default_payment_method': payment_method_id}
        )

        return payment_method

    @staticmethod
    async def list_payment_methods(customer_id: str, type: str = "card"):
        StripeClient._ensure_api_key()
        return stripe.PaymentMethod.list(
            customer=customer_id,
            type=type
        )

    @staticmethod
    async def create_subscription(
        customer_id: str,
        price_id: str,
        metadata: Dict = None
    ) -> stripe.Subscription:
        StripeClient._ensure_api_key()
        return stripe.Subscription.create(
            customer=customer_id,
            items=[{'price': price_id}],
            metadata=metadata or {}
        )

    @staticmethod
    async def cancel_subscription(subscription_id: str) -> stripe.Subscription:
        StripeClient._ensure_api_key()
        return stripe.Subscription.delete(subscription_id)