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