88 lines
2.8 KiB
Python
88 lines
2.8 KiB
Python
"""
|
|
app/billing/stripe_client.py
|
|
-----------------------------
|
|
Thin wrappers around the Stripe SDK. No Flask imports — usable from routes,
|
|
webhooks, and background tasks alike.
|
|
|
|
All functions raise RuntimeError with a clear message if STRIPE_SECRET_KEY is
|
|
absent, rather than producing cryptic Stripe SDK authentication errors.
|
|
"""
|
|
|
|
import os
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _get_stripe():
|
|
"""Import and initialise Stripe with the secret key. Raises if key is missing."""
|
|
import stripe as _stripe
|
|
key = os.environ.get('STRIPE_SECRET_KEY')
|
|
if not key:
|
|
raise RuntimeError(
|
|
'STRIPE_SECRET_KEY is not set. '
|
|
'Add it to your environment before enabling BILLING_ENABLED=true.'
|
|
)
|
|
_stripe.api_key = key
|
|
return _stripe
|
|
|
|
|
|
def create_stripe_customer(*, email: str, name: str, tenant_id: int, slug: str) -> dict:
|
|
"""Create a Stripe Customer and return the customer dict.
|
|
|
|
Stores tenant_id and slug in Stripe metadata for webhook reverse-lookup.
|
|
"""
|
|
stripe = _get_stripe()
|
|
customer = stripe.Customer.create(
|
|
email=email,
|
|
name=name,
|
|
metadata={'tenant_id': str(tenant_id), 'slug': slug},
|
|
)
|
|
logger.info('BILLING | stripe_customer_created | tenant=%s customer=%s', slug, customer['id'])
|
|
return customer
|
|
|
|
|
|
def create_checkout_session(
|
|
*,
|
|
customer_id: str,
|
|
price_id: str,
|
|
tenant_id: int,
|
|
success_url: str,
|
|
cancel_url: str,
|
|
) -> dict:
|
|
"""Create a Stripe Checkout Session for a subscription and return the session dict.
|
|
|
|
`success_url` should contain `{CHECKOUT_SESSION_ID}` which Stripe replaces
|
|
with the actual session ID on redirect (pass as `{{CHECKOUT_SESSION_ID}}`
|
|
inside an f-string to produce the literal braces).
|
|
"""
|
|
stripe = _get_stripe()
|
|
session = stripe.checkout.Session.create(
|
|
customer=customer_id,
|
|
mode='subscription',
|
|
line_items=[{'price': price_id, 'quantity': 1}],
|
|
success_url=success_url,
|
|
cancel_url=cancel_url,
|
|
metadata={'tenant_id': str(tenant_id)},
|
|
allow_promotion_codes=True,
|
|
)
|
|
logger.info('BILLING | checkout_session_created | tenant_id=%s session=%s', tenant_id, session['id'])
|
|
return session
|
|
|
|
|
|
def create_portal_session(*, customer_id: str, return_url: str) -> dict:
|
|
"""Create a Stripe Billing Portal Session and return the session dict."""
|
|
stripe = _get_stripe()
|
|
session = stripe.billing_portal.Session.create(
|
|
customer=customer_id,
|
|
return_url=return_url,
|
|
)
|
|
logger.info('BILLING | portal_session_created | customer=%s', customer_id)
|
|
return session
|
|
|
|
|
|
def retrieve_subscription(subscription_id: str) -> dict:
|
|
"""Retrieve a Stripe Subscription object."""
|
|
stripe = _get_stripe()
|
|
return stripe.Subscription.retrieve(subscription_id)
|