rbox/mywebdav/billing/stripe_client.py

import stripe
from typing import Dict
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)