sboulahtit/orion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(billing): end-to-end Stripe subscription signup with platform enforcement

Browse Source 
Move core signup service from marketplace to billing module, add
automatic Stripe product/price sync for tiers, create loyalty-specific
signup wizard, and enforce that platform is always explicitly known
(no silent defaulting to primary/hardcoded ID).

Key changes:
- New billing SignupService with separated account/store creation steps
- Stripe auto-sync on tier create/update (new prices, archive old)
- Loyalty signup template (Plan → Account → Store → Payment)
- platform_code is now required throughout the signup flow
- Pricing/signup pages return 404 if platform not detected
- OMS-specific logic (Letzshop claiming) stays in marketplace module
- Bootstrap script: scripts/seed/sync_stripe_products.py

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Samir Boulahtit
2026-02-28 19:16:14 +01:00
parent 2078ce35b2
commit 9684747d08
12 changed files with 1723 additions and 689 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
app/modules/billing/services/__init__.py
View File

@@ -21,6 +21,10 @@ from app.modules.billing.services.platform_pricing_service import (
PlatformPricingService,
platform_pricing_service,
)
from app.modules.billing.services.signup_service import (
SignupService,
signup_service,
)
from app.modules.billing.services.store_platform_sync_service import (
StorePlatformSync,
store_platform_sync,
@@ -65,4 +69,6 @@ __all__ = [
"TierInfoData",
"UpgradeTierData",
"LimitCheckData",
"SignupService",
"signup_service",
]

150
app/modules/billing/services/admin_subscription_service.py
View File

@@ -35,6 +35,141 @@ logger = logging.getLogger(__name__)
class AdminSubscriptionService:
"""Service for admin subscription management operations."""
# =========================================================================
# Stripe Tier Sync
# =========================================================================
@staticmethod
def _sync_tier_to_stripe(db: Session, tier: SubscriptionTier) -> None:
"""
Sync a tier's Stripe product and prices.
Creates or verifies the Stripe Product and Price objects, and
populates the stripe_product_id, stripe_price_monthly_id, and
stripe_price_annual_id fields on the tier.
Skips gracefully if Stripe is not configured (dev mode).
Stripe Prices are immutable — on price changes, new Prices are
created and old ones archived.
"""
from app.core.config import settings
if not settings.stripe_secret_key:
logger.debug(
f"Stripe not configured, skipping sync for tier {tier.code}"
)
return
import stripe
stripe.api_key = settings.stripe_secret_key
# Resolve platform name for product naming
platform_name = "Platform"
if tier.platform_id:
from app.modules.tenancy.services.platform_service import (
platform_service,
)
try:
platform = platform_service.get_platform_by_id(db, tier.platform_id)
platform_name = platform.name
except Exception:
pass
# --- Product ---
if tier.stripe_product_id:
# Verify it still exists in Stripe
try:
stripe.Product.retrieve(tier.stripe_product_id)
except stripe.InvalidRequestError:
logger.warning(
f"Stripe product {tier.stripe_product_id} not found, "
f"recreating for tier {tier.code}"
)
tier.stripe_product_id = None
if not tier.stripe_product_id:
product = stripe.Product.create(
name=f"{platform_name} - {tier.name}",
metadata={
"tier_code": tier.code,
"platform_id": str(tier.platform_id or ""),
},
)
tier.stripe_product_id = product.id
logger.info(
f"Created Stripe product {product.id} for tier {tier.code}"
)
# --- Monthly Price ---
if tier.price_monthly_cents:
if tier.stripe_price_monthly_id:
# Verify price matches; if not, create new one
try:
existing = stripe.Price.retrieve(tier.stripe_price_monthly_id)
if existing.unit_amount != tier.price_monthly_cents:
# Price changed — archive old, create new
stripe.Price.modify(
tier.stripe_price_monthly_id, active=False
)
tier.stripe_price_monthly_id = None
logger.info(
f"Archived old monthly price for tier {tier.code}"
)
except stripe.InvalidRequestError:
tier.stripe_price_monthly_id = None
if not tier.stripe_price_monthly_id:
price = stripe.Price.create(
product=tier.stripe_product_id,
unit_amount=tier.price_monthly_cents,
currency="eur",
recurring={"interval": "month"},
metadata={
"tier_code": tier.code,
"billing_period": "monthly",
},
)
tier.stripe_price_monthly_id = price.id
logger.info(
f"Created Stripe monthly price {price.id} "
f"for tier {tier.code} ({tier.price_monthly_cents} cents)"
)
# --- Annual Price ---
if tier.price_annual_cents:
if tier.stripe_price_annual_id:
try:
existing = stripe.Price.retrieve(tier.stripe_price_annual_id)
if existing.unit_amount != tier.price_annual_cents:
stripe.Price.modify(
tier.stripe_price_annual_id, active=False
)
tier.stripe_price_annual_id = None
logger.info(
f"Archived old annual price for tier {tier.code}"
)
except stripe.InvalidRequestError:
tier.stripe_price_annual_id = None
if not tier.stripe_price_annual_id:
price = stripe.Price.create(
product=tier.stripe_product_id,
unit_amount=tier.price_annual_cents,
currency="eur",
recurring={"interval": "year"},
metadata={
"tier_code": tier.code,
"billing_period": "annual",
},
)
tier.stripe_price_annual_id = price.id
logger.info(
f"Created Stripe annual price {price.id} "
f"for tier {tier.code} ({tier.price_annual_cents} cents)"
)
# =========================================================================
# Subscription Tiers
# =========================================================================
@@ -85,6 +220,9 @@ class AdminSubscriptionService:
tier = SubscriptionTier(**tier_data)
db.add(tier)
db.flush() # Get tier.id before Stripe sync
self._sync_tier_to_stripe(db, tier)
logger.info(f"Created subscription tier: {tier.code}")
return tier
@@ -95,9 +233,21 @@ class AdminSubscriptionService:
"""Update a subscription tier."""
tier = self.get_tier_by_code(db, tier_code)
# Track price changes to know if Stripe sync is needed
price_changed = (
"price_monthly_cents" in update_data
and update_data["price_monthly_cents"] != tier.price_monthly_cents
) or (
"price_annual_cents" in update_data
and update_data["price_annual_cents"] != tier.price_annual_cents
)
for field, value in update_data.items():
setattr(tier, field, value)
if price_changed or not tier.stripe_product_id:
self._sync_tier_to_stripe(db, tier)
logger.info(f"Updated subscription tier: {tier.code}")
return tier

797
app/modules/billing/services/signup_service.py Normal file
View File

@@ -0,0 +1,797 @@
# app/modules/billing/services/signup_service.py
"""
Core platform signup service.
Handles all database operations for the platform signup flow:
- Session management
- Account creation (User + Merchant)
- Store creation (separate step)
- Stripe customer & subscription setup
- Payment method collection
Platform-specific signup extensions (e.g., OMS Letzshop claiming)
live in their respective modules and call into this core service.
"""
from __future__ import annotations
import logging
import secrets
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta
from typing import TYPE_CHECKING
from sqlalchemy.orm import Session
from app.core.config import settings
from app.exceptions import (
ConflictException,
ResourceNotFoundException,
ValidationException,
)
from app.modules.billing.services.stripe_service import stripe_service
from app.modules.billing.services.subscription_service import (
subscription_service as sub_service,
)
from app.modules.messaging.services.email_service import EmailService
from middleware.auth import AuthManager
if TYPE_CHECKING:
from app.modules.tenancy.models import Store, User
logger = logging.getLogger(__name__)
# =============================================================================
# In-memory signup session storage
# In production, use Redis or database table
# =============================================================================
_signup_sessions: dict[str, dict] = {}
def _create_session_id() -> str:
"""Generate a secure session ID."""
return secrets.token_urlsafe(32)
# =============================================================================
# Data Classes
# =============================================================================
@dataclass
class SignupSessionData:
"""Data stored in a signup session."""
session_id: str
step: str
tier_code: str
is_annual: bool
platform_code: str = ""
created_at: str = ""
updated_at: str | None = None
store_name: str | None = None
user_id: int | None = None
merchant_id: int | None = None
store_id: int | None = None
store_code: str | None = None
platform_id: int | None = None
stripe_customer_id: str | None = None
setup_intent_id: str | None = None
extra: dict = field(default_factory=dict)
@dataclass
class AccountCreationResult:
"""Result of account creation."""
user_id: int
merchant_id: int
stripe_customer_id: str
@dataclass
class StoreCreationResult:
"""Result of store creation."""
store_id: int
store_code: str
@dataclass
class SignupCompletionResult:
"""Result of signup completion."""
success: bool
store_code: str
store_id: int
redirect_url: str
trial_ends_at: str
access_token: str | None = None # JWT token for automatic login
# =============================================================================
# Platform Signup Service
# =============================================================================
class SignupService:
"""Core service for handling platform signup operations."""
def __init__(self):
self.auth_manager = AuthManager()
# =========================================================================
# Session Management
# =========================================================================
def create_session(
self,
tier_code: str,
is_annual: bool,
platform_code: str,
) -> str:
"""
Create a new signup session.
Args:
tier_code: The subscription tier code
is_annual: Whether annual billing is selected
platform_code: Platform code (e.g., 'loyalty', 'oms')
Returns:
The session ID
Raises:
ValidationException: If tier code or platform code is invalid
"""
if not platform_code:
raise ValidationException(
message="Platform code is required for signup.",
field="platform_code",
)
# Validate tier code
from app.modules.billing.models import TierCode
try:
tier = TierCode(tier_code)
except ValueError:
raise ValidationException(
message=f"Invalid tier code: {tier_code}",
field="tier_code",
)
session_id = _create_session_id()
now = datetime.now(UTC).isoformat()
_signup_sessions[session_id] = {
"step": "tier_selected",
"tier_code": tier.value,
"is_annual": is_annual,
"platform_code": platform_code,
"created_at": now,
"updated_at": now,
}
logger.info(
f"Created signup session {session_id} for tier {tier.value}"
f" on platform {platform_code}"
)
return session_id
def get_session(self, session_id: str) -> dict | None:
"""Get a signup session by ID."""
return _signup_sessions.get(session_id)
def get_session_or_raise(self, session_id: str) -> dict:
"""
Get a signup session or raise an exception.
Raises:
ResourceNotFoundException: If session not found
"""
session = self.get_session(session_id)
if not session:
raise ResourceNotFoundException(
resource_type="SignupSession",
identifier=session_id,
)
return session
def update_session(self, session_id: str, data: dict) -> None:
"""Update signup session data."""
session = self.get_session_or_raise(session_id)
session.update(data)
session["updated_at"] = datetime.now(UTC).isoformat()
_signup_sessions[session_id] = session
def delete_session(self, session_id: str) -> None:
"""Delete a signup session."""
_signup_sessions.pop(session_id, None)
# =========================================================================
# Platform Resolution
# =========================================================================
def _resolve_platform_id(self, db: Session, session: dict) -> int:
"""
Resolve platform_id from session data.
The platform_code is always required in the session (set during
create_session). Raises if the platform cannot be resolved.
Raises:
ValidationException: If platform_code is missing or unknown
"""
from app.modules.tenancy.services.platform_service import platform_service
platform_code = session.get("platform_code")
if not platform_code:
raise ValidationException(
message="Platform code is missing from signup session.",
field="platform_code",
)
platform = platform_service.get_platform_by_code_optional(
db, platform_code
)
if not platform:
raise ValidationException(
message=f"Unknown platform: {platform_code}",
field="platform_code",
)
return platform.id
# =========================================================================
# Account Creation (User + Merchant only)
# =========================================================================
def check_email_exists(self, db: Session, email: str) -> bool:
"""Check if an email already exists."""
from app.modules.tenancy.services.admin_service import admin_service
return admin_service.get_user_by_email(db, email) is not None
def generate_unique_username(self, db: Session, email: str) -> str:
"""Generate a unique username from email."""
from app.modules.tenancy.services.admin_service import admin_service
username = email.split("@")[0]
base_username = username
counter = 1
while admin_service.get_user_by_username(db, username):
username = f"{base_username}_{counter}"
counter += 1
return username
def generate_unique_store_code(self, db: Session, name: str) -> str:
"""Generate a unique store code from a name."""
from app.modules.tenancy.services.store_service import store_service
store_code = name.upper().replace(" ", "_")[:20]
base_code = store_code
counter = 1
while store_service.is_store_code_taken(db, store_code):
store_code = f"{base_code}_{counter}"
counter += 1
return store_code
def generate_unique_subdomain(self, db: Session, name: str) -> str:
"""Generate a unique subdomain from a name."""
from app.modules.tenancy.services.store_service import store_service
subdomain = name.lower().replace(" ", "-")
subdomain = "".join(c for c in subdomain if c.isalnum() or c == "-")[:50]
base_subdomain = subdomain
counter = 1
while store_service.is_subdomain_taken(db, subdomain):
subdomain = f"{base_subdomain}-{counter}"
counter += 1
return subdomain
def create_account(
self,
db: Session,
session_id: str,
email: str,
password: str,
first_name: str,
last_name: str,
merchant_name: str,
phone: str | None = None,
) -> AccountCreationResult:
"""
Create user and merchant accounts.
Creates User + Merchant + Stripe Customer. Store creation is a
separate step (create_store) so each platform can customize it.
Args:
db: Database session
session_id: Signup session ID
email: User email
password: User password
first_name: User first name
last_name: User last name
merchant_name: Merchant/business name
phone: Optional phone number
Returns:
AccountCreationResult with user and merchant IDs
Raises:
ResourceNotFoundException: If session not found
ConflictException: If email already exists
"""
session = self.get_session_or_raise(session_id)
# Check if email already exists
if self.check_email_exists(db, email):
raise ConflictException(
message="An account with this email already exists",
)
# Generate unique username
username = self.generate_unique_username(db, email)
# Create User
from app.modules.tenancy.models import Merchant, User
user = User(
email=email,
username=username,
hashed_password=self.auth_manager.hash_password(password),
first_name=first_name,
last_name=last_name,
role="merchant_owner",
is_active=True,
)
db.add(user)
db.flush()
# Create Merchant
merchant = Merchant(
name=merchant_name,
owner_user_id=user.id,
contact_email=email,
contact_phone=phone,
)
db.add(merchant)
db.flush()
# Create Stripe Customer (linked to merchant, not store)
# We use a temporary store-like object for Stripe metadata
stripe_customer_id = stripe_service.create_customer_for_merchant(
merchant=merchant,
email=email,
name=f"{first_name} {last_name}",
metadata={
"merchant_name": merchant_name,
"tier": session.get("tier_code"),
"platform": session.get("platform_code", ""),
},
)
db.commit() # SVC-006 - Atomic account creation needs commit
# Update session
self.update_session(session_id, {
"user_id": user.id,
"merchant_id": merchant.id,
"merchant_name": merchant_name,
"email": email,
"stripe_customer_id": stripe_customer_id,
"step": "account_created",
})
logger.info(
f"Created account for {email}: "
f"user_id={user.id}, merchant_id={merchant.id}"
)
return AccountCreationResult(
user_id=user.id,
merchant_id=merchant.id,
stripe_customer_id=stripe_customer_id,
)
# =========================================================================
# Store Creation (separate step)
# =========================================================================
def create_store(
self,
db: Session,
session_id: str,
store_name: str | None = None,
language: str | None = None,
) -> StoreCreationResult:
"""
Create the first store for the merchant.
Store name defaults to the merchant name if not provided.
The merchant can modify store details later in the merchant panel.
Args:
db: Database session
session_id: Signup session ID
store_name: Store name (defaults to merchant name)
language: Store language code (e.g., 'fr', 'en', 'de')
Returns:
StoreCreationResult with store ID and code
Raises:
ResourceNotFoundException: If session not found
ValidationException: If account not created yet
"""
session = self.get_session_or_raise(session_id)
merchant_id = session.get("merchant_id")
if not merchant_id:
raise ValidationException(
message="Account not created. Please complete the account step first.",
field="session_id",
)
from app.modules.tenancy.models import Store
# Use merchant name as default store name
effective_name = store_name or session.get("merchant_name", "My Store")
email = session.get("email")
# Generate unique store code and subdomain
store_code = self.generate_unique_store_code(db, effective_name)
subdomain = self.generate_unique_subdomain(db, effective_name)
# Create Store
store = Store(
merchant_id=merchant_id,
store_code=store_code,
subdomain=subdomain,
name=effective_name,
contact_email=email,
is_active=True,
)
if language:
store.default_language = language
db.add(store)
db.flush()
# Resolve platform and create subscription
platform_id = self._resolve_platform_id(db, session)
# Create MerchantSubscription (trial status)
stripe_customer_id = session.get("stripe_customer_id")
subscription = sub_service.create_merchant_subscription(
db=db,
merchant_id=merchant_id,
platform_id=platform_id,
tier_code=session.get("tier_code", "essential"),
trial_days=settings.stripe_trial_days,
is_annual=session.get("is_annual", False),
)
subscription.stripe_customer_id = stripe_customer_id
db.commit() # SVC-006 - Atomic store creation needs commit
# Update session
self.update_session(session_id, {
"store_id": store.id,
"store_code": store_code,
"platform_id": platform_id,
"step": "store_created",
})
logger.info(
f"Created store {store_code} for merchant {merchant_id} "
f"on platform {session.get('platform_code')}"
)
return StoreCreationResult(
store_id=store.id,
store_code=store_code,
)
# =========================================================================
# Payment Setup
# =========================================================================
def setup_payment(self, session_id: str) -> tuple[str, str]:
"""
Create Stripe SetupIntent for card collection.
Args:
session_id: Signup session ID
Returns:
Tuple of (client_secret, stripe_customer_id)
Raises:
ResourceNotFoundException: If session not found
ValidationException: If store not created yet
"""
session = self.get_session_or_raise(session_id)
stripe_customer_id = session.get("stripe_customer_id")
if not stripe_customer_id:
raise ValidationException(
message="Account not created. Please complete earlier steps first.",
field="session_id",
)
if not session.get("store_id"):
raise ValidationException(
message="Store not created. Please complete the store step first.",
field="session_id",
)
# Create SetupIntent
setup_intent = stripe_service.create_setup_intent(
customer_id=stripe_customer_id,
metadata={
"session_id": session_id,
"store_id": str(session.get("store_id")),
"tier": session.get("tier_code"),
},
)
# Update session
self.update_session(session_id, {
"setup_intent_id": setup_intent.id,
"step": "payment_pending",
})
logger.info(
f"Created SetupIntent {setup_intent.id} for session {session_id}"
)
return setup_intent.client_secret, stripe_customer_id
# =========================================================================
# Welcome Email
# =========================================================================
def send_welcome_email(
self,
db: Session,
user: User,
store: Store,
tier_code: str,
language: str = "fr",
) -> None:
"""
Send welcome email to new store.
Args:
db: Database session
user: User who signed up
store: Store that was created
tier_code: Selected tier code
language: Language for email (default: French)
"""
try:
# Get tier name
from app.modules.billing.services.billing_service import billing_service
tier = billing_service.get_tier_by_code(db, tier_code)
tier_name = tier.name if tier else tier_code.title()
# Build login URL
login_url = (
f"https://{settings.platform_domain}"
f"/store/{store.store_code}/dashboard"
)
email_service = EmailService(db)
email_service.send_template(
template_code="signup_welcome",
language=language,
to_email=user.email,
to_name=f"{user.first_name} {user.last_name}",
variables={
"first_name": user.first_name,
"merchant_name": store.name,
"email": user.email,
"store_code": store.store_code,
"login_url": login_url,
"trial_days": settings.stripe_trial_days,
"tier_name": tier_name,
},
store_id=store.id,
user_id=user.id,
related_type="signup",
)
logger.info(f"Welcome email sent to {user.email}")
except Exception as e: # noqa: EXC003
# Log error but don't fail signup
logger.error(f"Failed to send welcome email to {user.email}: {e}")
# =========================================================================
# Signup Completion
# =========================================================================
def complete_signup(
self,
db: Session,
session_id: str,
setup_intent_id: str,
) -> SignupCompletionResult:
"""
Complete signup after card collection.
Verifies the SetupIntent, attaches the payment method to the Stripe
customer, creates the Stripe Subscription with trial, and generates
a JWT token for automatic login.
Args:
db: Database session
session_id: Signup session ID
setup_intent_id: Stripe SetupIntent ID
Returns:
SignupCompletionResult
Raises:
ResourceNotFoundException: If session not found
ValidationException: If signup incomplete or payment failed
"""
from app.modules.billing.models import SubscriptionTier, TierCode
session = self.get_session_or_raise(session_id)
# Guard against completing signup more than once
if session.get("step") == "completed":
raise ValidationException(
message="Signup already completed.",
field="session_id",
)
store_id = session.get("store_id")
stripe_customer_id = session.get("stripe_customer_id")
if not store_id or not stripe_customer_id:
raise ValidationException(
message="Incomplete signup. Please start again.",
field="session_id",
)
# Retrieve SetupIntent to get payment method
setup_intent = stripe_service.get_setup_intent(setup_intent_id)
if setup_intent.status != "succeeded":
raise ValidationException(
message="Card setup not completed. Please try again.",
field="setup_intent_id",
)
payment_method_id = setup_intent.payment_method
# Attach payment method to customer
stripe_service.attach_payment_method_to_customer(
customer_id=stripe_customer_id,
payment_method_id=payment_method_id,
set_as_default=True,
)
# Update subscription record
subscription = sub_service.get_subscription_for_store(db, store_id)
if subscription:
subscription.stripe_payment_method_id = payment_method_id
# Create the actual Stripe Subscription with trial period
# This is what enables automatic charging after trial ends
if subscription.tier_id:
tier = (
db.query(SubscriptionTier)
.filter(SubscriptionTier.id == subscription.tier_id)
.first()
)
if tier:
price_id = (
tier.stripe_price_annual_id
if subscription.is_annual and tier.stripe_price_annual_id
else tier.stripe_price_monthly_id
)
if price_id:
stripe_sub = stripe_service.create_subscription_with_trial(
customer_id=stripe_customer_id,
price_id=price_id,
trial_days=settings.stripe_trial_days,
metadata={
"merchant_id": str(subscription.merchant_id),
"platform_id": str(subscription.platform_id),
"tier_code": tier.code,
},
)
subscription.stripe_subscription_id = stripe_sub.id
logger.info(
f"Created Stripe subscription {stripe_sub.id} "
f"for merchant {subscription.merchant_id}"
)
db.commit() # SVC-006 - Finalize signup needs commit
# Get store info
from app.modules.tenancy.models import Store, User
store = db.query(Store).filter(Store.id == store_id).first()
store_code = store.store_code if store else session.get("store_code")
trial_ends_at = (
subscription.trial_ends_at
if subscription
else datetime.now(UTC) + timedelta(days=30)
)
# Get user for welcome email and token generation
user_id = session.get("user_id")
user = (
db.query(User).filter(User.id == user_id).first() if user_id else None
)
# Generate access token for automatic login after signup
access_token = None
if user and store:
# Create store-scoped JWT token (user is owner since they just signed up)
token_data = self.auth_manager.create_access_token(
user=user,
store_id=store.id,
store_code=store.store_code,
store_role="Owner", # New signup is always the owner
)
access_token = token_data["access_token"]
logger.info(f"Generated access token for new store user {user.email}")
# Send welcome email
if user and store:
tier_code = session.get("tier_code", TierCode.ESSENTIAL.value)
self.send_welcome_email(db, user, store, tier_code)
# Determine redirect based on platform
redirect_url = self._get_post_signup_redirect(db, session, store_code)
# Clean up session
self.delete_session(session_id)
logger.info(f"Completed signup for store {store_id}")
return SignupCompletionResult(
success=True,
store_code=store_code,
store_id=store_id,
redirect_url=redirect_url,
trial_ends_at=trial_ends_at.isoformat(),
access_token=access_token,
)
def _get_post_signup_redirect(
self, db: Session, session: dict, store_code: str
) -> str:
"""
Determine redirect URL after signup based on platform.
Marketplace platforms → onboarding wizard.
Other platforms (loyalty, etc.) → dashboard.
"""
from app.modules.service import module_service
platform_id = session.get("platform_id")
if platform_id:
try:
if module_service.is_module_enabled(db, platform_id, "marketplace"):
return f"/store/{store_code}/onboarding"
except Exception:
pass # If check fails, default to dashboard
return f"/store/{store_code}/dashboard"
# Singleton instance
signup_service = SignupService()

32
app/modules/billing/services/stripe_service.py
View File

@@ -93,6 +93,38 @@ class StripeService:
)
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()