|
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)
|