Samir Boulahtit
319900623a
Some checks failed
- Add admin SQL query tool with saved queries, schema explorer presets, and collapsible category sections (dev_tools module) - Add platform debug tool for admin diagnostics - Add loyalty settings page with owner-only access control - Fix loyalty settings owner check (use currentUser instead of window.__userData) - Replace HTTPException with AuthorizationException in loyalty routes - Expand loyalty module with PIN service, Apple Wallet, program management - Improve store login with platform detection and multi-platform support - Update billing feature gates and subscription services - Add store platform sync improvements and remove is_primary column - Add unit tests for loyalty (PIN, points, stamps, program services) - Update i18n translations across dev_tools locales Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
620 lines
19 KiB
Python
620 lines
19 KiB
Python
# app/modules/billing/services/stripe_service.py
|
|
"""
|
|
Stripe payment integration service.
|
|
|
|
Provides:
|
|
- Customer management
|
|
- Subscription management
|
|
- Checkout session creation
|
|
- Customer portal access
|
|
- Webhook event construction
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import TYPE_CHECKING
|
|
|
|
import stripe
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.config import settings
|
|
from app.modules.billing.exceptions import (
|
|
StripeNotConfiguredException,
|
|
WebhookVerificationException,
|
|
)
|
|
from app.modules.billing.models import (
|
|
MerchantSubscription,
|
|
)
|
|
|
|
if TYPE_CHECKING:
|
|
from app.modules.tenancy.models import Store
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class StripeService:
|
|
"""Service for Stripe payment operations."""
|
|
|
|
def __init__(self):
|
|
self._configured = False
|
|
self._configure()
|
|
|
|
def _configure(self):
|
|
"""Configure Stripe with API key."""
|
|
if settings.stripe_secret_key:
|
|
stripe.api_key = settings.stripe_secret_key
|
|
self._configured = True
|
|
else:
|
|
logger.debug("Stripe API key not configured")
|
|
|
|
@property
|
|
def is_configured(self) -> bool:
|
|
"""Check if Stripe is properly configured."""
|
|
return self._configured and bool(settings.stripe_secret_key)
|
|
|
|
def _check_configured(self) -> None:
|
|
"""Raise exception if Stripe is not configured."""
|
|
if not self.is_configured:
|
|
raise StripeNotConfiguredException()
|
|
|
|
# =========================================================================
|
|
# Customer Management
|
|
# =========================================================================
|
|
|
|
def create_customer(
|
|
self,
|
|
store: Store,
|
|
email: str,
|
|
name: str | None = None,
|
|
metadata: dict | None = None,
|
|
) -> str:
|
|
"""
|
|
Create a Stripe customer for a store.
|
|
|
|
Returns the Stripe customer ID.
|
|
"""
|
|
self._check_configured()
|
|
|
|
customer_metadata = {
|
|
"store_id": str(store.id),
|
|
"store_code": store.store_code,
|
|
**(metadata or {}),
|
|
}
|
|
|
|
customer = stripe.Customer.create(
|
|
email=email,
|
|
name=name or store.name,
|
|
metadata=customer_metadata,
|
|
)
|
|
|
|
logger.info(
|
|
f"Created Stripe customer {customer.id} for store {store.store_code}"
|
|
)
|
|
return customer.id
|
|
|
|
def create_customer_for_merchant(
|
|
self,
|
|
merchant,
|
|
email: str,
|
|
name: str | None = None,
|
|
metadata: dict | None = None,
|
|
) -> str:
|
|
"""
|
|
Create a Stripe customer for a merchant (before store exists).
|
|
|
|
Used during signup when the store hasn't been created yet.
|
|
Returns the Stripe customer ID.
|
|
"""
|
|
self._check_configured()
|
|
|
|
customer_metadata = {
|
|
"merchant_id": str(merchant.id),
|
|
"merchant_name": merchant.name,
|
|
**(metadata or {}),
|
|
}
|
|
|
|
customer = stripe.Customer.create(
|
|
email=email,
|
|
name=name or merchant.name,
|
|
metadata=customer_metadata,
|
|
)
|
|
|
|
logger.info(
|
|
f"Created Stripe customer {customer.id} for merchant {merchant.name}"
|
|
)
|
|
return customer.id
|
|
|
|
def get_customer(self, customer_id: str) -> stripe.Customer:
|
|
"""Get a Stripe customer by ID."""
|
|
self._check_configured()
|
|
|
|
return stripe.Customer.retrieve(customer_id)
|
|
|
|
def update_customer(
|
|
self,
|
|
customer_id: str,
|
|
email: str | None = None,
|
|
name: str | None = None,
|
|
metadata: dict | None = None,
|
|
) -> stripe.Customer:
|
|
"""Update a Stripe customer."""
|
|
self._check_configured()
|
|
|
|
update_data = {}
|
|
if email:
|
|
update_data["email"] = email
|
|
if name:
|
|
update_data["name"] = name
|
|
if metadata:
|
|
update_data["metadata"] = metadata
|
|
|
|
return stripe.Customer.modify(customer_id, **update_data)
|
|
|
|
# =========================================================================
|
|
# Subscription Management
|
|
# =========================================================================
|
|
|
|
def create_subscription(
|
|
self,
|
|
customer_id: str,
|
|
price_id: str,
|
|
trial_days: int | None = None,
|
|
metadata: dict | None = None,
|
|
) -> stripe.Subscription:
|
|
"""
|
|
Create a new Stripe subscription.
|
|
|
|
Args:
|
|
customer_id: Stripe customer ID
|
|
price_id: Stripe price ID for the subscription
|
|
trial_days: Optional trial period in days
|
|
metadata: Optional metadata to attach
|
|
|
|
Returns:
|
|
Stripe Subscription object
|
|
"""
|
|
self._check_configured()
|
|
|
|
subscription_data = {
|
|
"customer": customer_id,
|
|
"items": [{"price": price_id}],
|
|
"metadata": metadata or {},
|
|
"payment_behavior": "default_incomplete",
|
|
"expand": ["latest_invoice.payment_intent"],
|
|
}
|
|
|
|
if trial_days:
|
|
subscription_data["trial_period_days"] = trial_days
|
|
|
|
subscription = stripe.Subscription.create(**subscription_data)
|
|
logger.info(
|
|
f"Created Stripe subscription {subscription.id} for customer {customer_id}"
|
|
)
|
|
return subscription
|
|
|
|
def get_subscription(self, subscription_id: str) -> stripe.Subscription:
|
|
"""Get a Stripe subscription by ID."""
|
|
self._check_configured()
|
|
|
|
return stripe.Subscription.retrieve(subscription_id)
|
|
|
|
def update_subscription(
|
|
self,
|
|
subscription_id: str,
|
|
new_price_id: str | None = None,
|
|
proration_behavior: str = "create_prorations",
|
|
metadata: dict | None = None,
|
|
) -> stripe.Subscription:
|
|
"""
|
|
Update a Stripe subscription (e.g., change tier).
|
|
|
|
Args:
|
|
subscription_id: Stripe subscription ID
|
|
new_price_id: New price ID for tier change
|
|
proration_behavior: How to handle prorations
|
|
metadata: Optional metadata to update
|
|
|
|
Returns:
|
|
Updated Stripe Subscription object
|
|
"""
|
|
self._check_configured()
|
|
|
|
update_data = {"proration_behavior": proration_behavior}
|
|
|
|
if new_price_id:
|
|
# Get the subscription to find the item ID
|
|
subscription = stripe.Subscription.retrieve(subscription_id)
|
|
item_id = subscription["items"]["data"][0]["id"]
|
|
update_data["items"] = [{"id": item_id, "price": new_price_id}]
|
|
|
|
if metadata:
|
|
update_data["metadata"] = metadata
|
|
|
|
updated = stripe.Subscription.modify(subscription_id, **update_data)
|
|
logger.info(f"Updated Stripe subscription {subscription_id}")
|
|
return updated
|
|
|
|
def cancel_subscription(
|
|
self,
|
|
subscription_id: str,
|
|
immediately: bool = False,
|
|
cancellation_reason: str | None = None,
|
|
) -> stripe.Subscription:
|
|
"""
|
|
Cancel a Stripe subscription.
|
|
|
|
Args:
|
|
subscription_id: Stripe subscription ID
|
|
immediately: If True, cancel now. If False, cancel at period end.
|
|
cancellation_reason: Optional reason for cancellation
|
|
|
|
Returns:
|
|
Cancelled Stripe Subscription object
|
|
"""
|
|
self._check_configured()
|
|
|
|
if immediately:
|
|
subscription = stripe.Subscription.cancel(subscription_id)
|
|
else:
|
|
subscription = stripe.Subscription.modify(
|
|
subscription_id,
|
|
cancel_at_period_end=True,
|
|
metadata={"cancellation_reason": cancellation_reason or "user_request"},
|
|
)
|
|
|
|
logger.info(
|
|
f"Cancelled Stripe subscription {subscription_id} "
|
|
f"(immediately={immediately})"
|
|
)
|
|
return subscription
|
|
|
|
def reactivate_subscription(self, subscription_id: str) -> stripe.Subscription:
|
|
"""
|
|
Reactivate a cancelled subscription (if not yet ended).
|
|
|
|
Returns:
|
|
Reactivated Stripe Subscription object
|
|
"""
|
|
self._check_configured()
|
|
|
|
subscription = stripe.Subscription.modify(
|
|
subscription_id,
|
|
cancel_at_period_end=False,
|
|
)
|
|
logger.info(f"Reactivated Stripe subscription {subscription_id}")
|
|
return subscription
|
|
|
|
def cancel_subscription_item(self, subscription_item_id: str) -> None:
|
|
"""
|
|
Cancel a subscription item (used for add-ons).
|
|
|
|
Args:
|
|
subscription_item_id: Stripe subscription item ID
|
|
"""
|
|
self._check_configured()
|
|
|
|
stripe.SubscriptionItem.delete(subscription_item_id)
|
|
logger.info(f"Cancelled Stripe subscription item {subscription_item_id}")
|
|
|
|
# =========================================================================
|
|
# Checkout & Portal
|
|
# =========================================================================
|
|
|
|
def create_checkout_session(
|
|
self,
|
|
db: Session,
|
|
store: Store,
|
|
price_id: str,
|
|
success_url: str,
|
|
cancel_url: str,
|
|
trial_days: int | None = None,
|
|
quantity: int = 1,
|
|
metadata: dict | None = None,
|
|
platform_id: int | None = None,
|
|
) -> stripe.checkout.Session:
|
|
"""
|
|
Create a Stripe Checkout session for subscription signup.
|
|
|
|
Args:
|
|
db: Database session
|
|
store: Store to create checkout for
|
|
price_id: Stripe price ID
|
|
success_url: URL to redirect on success
|
|
cancel_url: URL to redirect on cancel
|
|
trial_days: Optional trial period
|
|
quantity: Number of items (default 1)
|
|
metadata: Additional metadata to store
|
|
platform_id: Platform ID (from JWT or caller). Falls back to DB lookup.
|
|
|
|
Returns:
|
|
Stripe Checkout Session object
|
|
"""
|
|
self._check_configured()
|
|
|
|
# Get or create Stripe customer
|
|
from app.modules.tenancy.services.platform_service import platform_service
|
|
from app.modules.tenancy.services.team_service import team_service
|
|
|
|
if platform_id is None:
|
|
platform_id = platform_service.get_first_active_platform_id_for_store(db, store.id)
|
|
subscription = None
|
|
if store.merchant_id and platform_id:
|
|
subscription = (
|
|
db.query(MerchantSubscription)
|
|
.filter(
|
|
MerchantSubscription.merchant_id == store.merchant_id,
|
|
MerchantSubscription.platform_id == platform_id,
|
|
)
|
|
.first()
|
|
)
|
|
|
|
if subscription and subscription.stripe_customer_id:
|
|
customer_id = subscription.stripe_customer_id
|
|
else:
|
|
# Get store owner email
|
|
owner = team_service.get_store_owner(db, store.id)
|
|
email = owner.user.email if owner and owner.user else None
|
|
|
|
customer_id = self.create_customer(store, email or f"{store.store_code}@placeholder.com")
|
|
|
|
# Store the customer ID
|
|
if subscription:
|
|
subscription.stripe_customer_id = customer_id
|
|
db.flush()
|
|
|
|
# Build metadata
|
|
session_metadata = {
|
|
"store_id": str(store.id),
|
|
"store_code": store.store_code,
|
|
"merchant_id": str(store.merchant_id) if store.merchant_id else "",
|
|
}
|
|
if metadata:
|
|
session_metadata.update(metadata)
|
|
|
|
session_data = {
|
|
"customer": customer_id,
|
|
"line_items": [{"price": price_id, "quantity": quantity}],
|
|
"mode": "subscription",
|
|
"success_url": success_url,
|
|
"cancel_url": cancel_url,
|
|
"metadata": session_metadata,
|
|
}
|
|
|
|
if trial_days:
|
|
session_data["subscription_data"] = {"trial_period_days": trial_days}
|
|
|
|
session = stripe.checkout.Session.create(**session_data)
|
|
logger.info(f"Created checkout session {session.id} for store {store.store_code}")
|
|
return session
|
|
|
|
def create_portal_session(
|
|
self,
|
|
customer_id: str,
|
|
return_url: str,
|
|
) -> stripe.billing_portal.Session:
|
|
"""
|
|
Create a Stripe Customer Portal session.
|
|
|
|
Allows customers to manage their subscription, payment methods, and invoices.
|
|
|
|
Args:
|
|
customer_id: Stripe customer ID
|
|
return_url: URL to return to after portal
|
|
|
|
Returns:
|
|
Stripe Portal Session object
|
|
"""
|
|
self._check_configured()
|
|
|
|
session = stripe.billing_portal.Session.create(
|
|
customer=customer_id,
|
|
return_url=return_url,
|
|
)
|
|
logger.info(f"Created portal session for customer {customer_id}")
|
|
return session
|
|
|
|
# =========================================================================
|
|
# Invoice Management
|
|
# =========================================================================
|
|
|
|
def get_invoices(
|
|
self,
|
|
customer_id: str,
|
|
limit: int = 10,
|
|
) -> list[stripe.Invoice]:
|
|
"""Get invoices for a customer."""
|
|
self._check_configured()
|
|
|
|
invoices = stripe.Invoice.list(customer=customer_id, limit=limit)
|
|
return list(invoices.data)
|
|
|
|
def get_upcoming_invoice(self, customer_id: str) -> stripe.Invoice | None:
|
|
"""Get the upcoming invoice for a customer."""
|
|
self._check_configured()
|
|
|
|
try:
|
|
return stripe.Invoice.upcoming(customer=customer_id)
|
|
except stripe.error.InvalidRequestError:
|
|
# No upcoming invoice
|
|
return None
|
|
|
|
# =========================================================================
|
|
# Webhook Handling
|
|
# =========================================================================
|
|
|
|
def construct_event(
|
|
self,
|
|
payload: bytes,
|
|
sig_header: str,
|
|
) -> stripe.Event:
|
|
"""
|
|
Construct and verify a Stripe webhook event.
|
|
|
|
Args:
|
|
payload: Raw request body
|
|
sig_header: Stripe-Signature header
|
|
|
|
Returns:
|
|
Verified Stripe Event object
|
|
|
|
Raises:
|
|
WebhookVerificationException: If signature verification fails
|
|
"""
|
|
if not settings.stripe_webhook_secret:
|
|
raise WebhookVerificationException("Stripe webhook secret not configured")
|
|
|
|
try:
|
|
event = stripe.Webhook.construct_event(
|
|
payload,
|
|
sig_header,
|
|
settings.stripe_webhook_secret,
|
|
)
|
|
return event
|
|
except stripe.error.SignatureVerificationError as e:
|
|
logger.error(f"Webhook signature verification failed: {e}")
|
|
raise WebhookVerificationException("Invalid webhook signature")
|
|
|
|
# =========================================================================
|
|
# SetupIntent & Payment Method Management
|
|
# =========================================================================
|
|
|
|
def create_setup_intent(
|
|
self,
|
|
customer_id: str,
|
|
metadata: dict | None = None,
|
|
) -> stripe.SetupIntent:
|
|
"""
|
|
Create a SetupIntent to collect card without charging.
|
|
|
|
Used for trial signups where we collect card upfront
|
|
but don't charge until trial ends.
|
|
|
|
Args:
|
|
customer_id: Stripe customer ID
|
|
metadata: Optional metadata to attach
|
|
|
|
Returns:
|
|
Stripe SetupIntent object with client_secret for frontend
|
|
"""
|
|
self._check_configured()
|
|
|
|
setup_intent = stripe.SetupIntent.create(
|
|
customer=customer_id,
|
|
payment_method_types=["card"],
|
|
metadata=metadata or {},
|
|
)
|
|
|
|
logger.info(f"Created SetupIntent {setup_intent.id} for customer {customer_id}")
|
|
return setup_intent
|
|
|
|
def attach_payment_method_to_customer(
|
|
self,
|
|
customer_id: str,
|
|
payment_method_id: str,
|
|
set_as_default: bool = True,
|
|
) -> None:
|
|
"""
|
|
Attach a payment method to customer and optionally set as default.
|
|
|
|
Args:
|
|
customer_id: Stripe customer ID
|
|
payment_method_id: Payment method ID from confirmed SetupIntent
|
|
set_as_default: Whether to set as default payment method
|
|
"""
|
|
self._check_configured()
|
|
|
|
# Attach the payment method to the customer
|
|
stripe.PaymentMethod.attach(payment_method_id, customer=customer_id)
|
|
|
|
if set_as_default:
|
|
stripe.Customer.modify(
|
|
customer_id,
|
|
invoice_settings={"default_payment_method": payment_method_id},
|
|
)
|
|
|
|
logger.info(
|
|
f"Attached payment method {payment_method_id} to customer {customer_id} "
|
|
f"(default={set_as_default})"
|
|
)
|
|
|
|
def create_subscription_with_trial(
|
|
self,
|
|
customer_id: str,
|
|
price_id: str,
|
|
trial_days: int = 30,
|
|
metadata: dict | None = None,
|
|
) -> stripe.Subscription:
|
|
"""
|
|
Create subscription with trial period.
|
|
|
|
Customer must have a default payment method attached.
|
|
Card will be charged automatically after trial ends.
|
|
|
|
Args:
|
|
customer_id: Stripe customer ID (must have default payment method)
|
|
price_id: Stripe price ID for the subscription tier
|
|
trial_days: Number of trial days (default 30)
|
|
metadata: Optional metadata to attach
|
|
|
|
Returns:
|
|
Stripe Subscription object
|
|
"""
|
|
self._check_configured()
|
|
|
|
subscription = stripe.Subscription.create(
|
|
customer=customer_id,
|
|
items=[{"price": price_id}],
|
|
trial_period_days=trial_days,
|
|
metadata=metadata or {},
|
|
# Use default payment method for future charges
|
|
default_payment_method=None, # Uses customer's default
|
|
)
|
|
|
|
logger.info(
|
|
f"Created subscription {subscription.id} with {trial_days}-day trial "
|
|
f"for customer {customer_id}"
|
|
)
|
|
return subscription
|
|
|
|
def get_setup_intent(self, setup_intent_id: str) -> stripe.SetupIntent:
|
|
"""Get a SetupIntent by ID."""
|
|
self._check_configured()
|
|
|
|
return stripe.SetupIntent.retrieve(setup_intent_id)
|
|
|
|
# =========================================================================
|
|
# Price/Product Management
|
|
# =========================================================================
|
|
|
|
def get_price(self, price_id: str) -> stripe.Price:
|
|
"""Get a Stripe price by ID."""
|
|
self._check_configured()
|
|
|
|
return stripe.Price.retrieve(price_id)
|
|
|
|
def get_product(self, product_id: str) -> stripe.Product:
|
|
"""Get a Stripe product by ID."""
|
|
self._check_configured()
|
|
|
|
return stripe.Product.retrieve(product_id)
|
|
|
|
def list_prices(
|
|
self,
|
|
product_id: str | None = None,
|
|
active: bool = True,
|
|
) -> list[stripe.Price]:
|
|
"""List Stripe prices, optionally filtered by product."""
|
|
self._check_configured()
|
|
|
|
params = {"active": active}
|
|
if product_id:
|
|
params["product"] = product_id
|
|
|
|
prices = stripe.Price.list(**params)
|
|
return list(prices.data)
|
|
|
|
|
|
# Create service instance
|
|
stripe_service = StripeService()
|