class SharedPaymentMethodModel:
"""Share payment methods from platform customer to connected accounts"""
def attach_platform_payment_method_to_account(self,
user_id: str,
connected_account_id: str,
payment_method_id: str):
"""Clone payment method to connected account"""
try:
# Get payment method details from platform
pm = self.client.v1.payment_methods.retrieve(payment_method_id)
# Create equivalent payment method on connected account
if pm.type == "card":
# Note: This requires raw card data access OR use setup intents
connected_pm = self.create_equivalent_payment_method(
pm, connected_account_id
)
return connected_pm.id
except Exception as e:
print(f"Failed to share payment method: {e}")
return None
def process_payment_with_shared_method(self,
user_id: str,
connected_account_id: str,
amount: int,
platform_payment_method_id: str):
"""Process payment using shared payment method"""
# Get or create linked customer
linked_customer_id = self.link_to_connected_account(user_id, connected_account_id)
# Clone payment method to connected account if needed
connected_pm_id = self.get_or_clone_payment_method(
platform_payment_method_id, connected_account_id
)
# Process payment on connected account
intent = self.client.v1.payment_intents.create(
params={
"amount": amount,
"currency": "usd",
"customer": linked_customer_id,
"payment_method": connected_pm_id,
"confirm": True
},
options={"stripe_account": connected_account_id}
)
return intent
class SharedPaymentMethodModel:
"""Share payment methods from platform customer to connected accounts"""
def attach_platform_payment_method_to_account(self,
user_id: str,
connected_account_id: str,
payment_method_id: str):
"""Clone payment method to connected account"""
try:
# Get payment method details from platform
pm = self.client.v1.payment_methods.retrieve(payment_method_id)
# Create equivalent payment method on connected account
if pm.type == "card":
# Note: This requires raw card data access OR use setup intents
connected_pm = self.create_equivalent_payment_method(
pm, connected_account_id
)
return connected_pm.id
except Exception as e:
print(f"Failed to share payment method: {e}")
return None
def process_payment_with_shared_method(self,
user_id: str,
connected_account_id: str,
amount: int,
platform_payment_method_id: str):
"""Process payment using shared payment method"""
# Get or create linked customer
linked_customer_id = self.link_to_connected_account(user_id, connected_account_id)
# Clone payment method to connected account if needed
connected_pm_id = self.get_or_clone_payment_method(
platform_payment_method_id, connected_account_id
)
# Process payment on connected account
intent = self.client.v1.payment_intents.create(
params={
"amount": amount,
"currency": "usd",
"customer": linked_customer_id,
"payment_method": connected_pm_id,
"confirm": True
},
options={"stripe_account": connected_account_id}
)
return intent
class SetupIntentLinkingModel:
"""Use Setup Intents to link payment methods on demand"""
def create_setup_intent_for_account(self,
user_id: str,
connected_account_id: str):
"""Create setup intent on connected account for payment method collection"""
# Get or create linked customer
linked_customer_id = self.link_to_connected_account(user_id, connected_account_id)
# Create setup intent on connected account
setup_intent = self.client.v1.setup_intents.create(
params={
"customer": linked_customer_id,
"payment_method_types": ["card"],
"usage": "off_session",
"metadata": {
"platform_user_id": user_id,
"source": "on_demand_linking"
}
},
options={"stripe_account": connected_account_id}
)
return setup_intent.client_secret
def link_existing_platform_payment_method(self,
user_id: str,
connected_account_id: str,
platform_payment_method_id: str):
"""Link existing platform payment method to connected account"""
# This requires the customer to re-authorize their payment method
# for the connected account due to Stripe's security requirements
setup_intent_secret = self.create_setup_intent_for_account(
user_id, connected_account_id
)
return {
"requires_reauthorization": True,
"setup_intent_client_secret": setup_intent_secret,
"message": "Customer needs to re-authorize payment method for this merchant"
}
class LinkedCustomerModel:
"""Platform customer with on-demand connected account linking"""
def create_platform_customer(self, user_id: str, user_data: dict):
"""Create main customer on platform account"""
# Create customer on platform (no stripe_account header)
platform_customer = self.client.v1.customers.create(
params={
"name": user_data.get("name"),
"email": user_data.get("email"),
"phone": user_data.get("phone"),
"metadata": {
"platform_user_id": user_id,
"is_platform_master": "true"
}
}
)
# Store in database
self.db.store_platform_customer(user_id, platform_customer.id)
return platform_customer
def link_to_connected_account(self, user_id: str, connected_account_id: str):
"""Create linked customer on connected account when needed"""
# Get platform customer data
platform_customer_id = self.db.get_platform_customer_id(user_id)
platform_customer = self.client.v1.customers.retrieve(platform_customer_id)
# Check if already linked
existing_link = self.db.get_linked_customer(user_id, connected_account_id)
if existing_link:
return existing_link
# Create customer on connected account with same data
linked_customer = self.client.v1.customers.create(
params={
"name": platform_customer.name,
"email": platform_customer.email,
"phone": platform_customer.phone,
"metadata": {
**platform_customer.metadata,
"platform_customer_id": platform_customer_id,
"linked_from_platform": "true"
}
},
options={"stripe_account": connected_account_id}
)
# Store the link
self.db.store_customer_link(
user_id=user_id,
platform_customer_id=platform_customer_id,
connected_account_id=connected_account_id,
linked_customer_id=linked_customer.id
)
return linked_customer.id
def sync_customer_data(self, user_id: str, updated_data: dict):
"""Update customer data across all linked accounts"""
platform_customer_id = self.db.get_platform_customer_id(user_id)
# Update platform customer
self.client.v1.customers.modify(
platform_customer_id,
params=updated_data
)
# Update all linked customers
linked_customers = self.db.get_all_linked_customers(user_id)
for link in linked_customers:
try:
self.client.v1.customers.modify(
link["linked_customer_id"],
params=updated_data,
options={"stripe_account": link["connected_account_id"]}
)
except Exception as e:
print(f"Failed to sync to {link['connected_account_id']}: {e}")
def charge_with_manual_capture(self,
customer_id: str,
amount: int,
connected_account_id: str,
delay_days: int,
payment_method_id: str):
"""For delays ≤ 7 days: Use manual capture"""
# Create PaymentIntent with manual capture
intent = self.client.v1.payment_intents.create(
params={
"amount": amount,
"currency": "usd",
"customer": customer_id,
"payment_method": payment_method_id,
"capture_method": "manual",
"payment_method_options": {
"card": {
"request_extended_authorization": "if_available"
}
},
"metadata": {
"delay_days": str(delay_days),
"capture_after": str(int(time.time()) + (delay_days * 24 * 60 * 60)),
"risk_review": "pending"
}
},
options={
"stripe_account": connected_account_id,
**self.req_options
}
)
# Confirm the payment (authorize funds)
confirmed_intent = self.client.v1.payment_intents.confirm(
intent.id,
options={
"stripe_account": connected_account_id,
**self.req_options
}
)
# Schedule capture after delay period
self.schedule_delayed_capture(
payment_intent_id=confirmed_intent.id,
connected_account_id=connected_account_id,
delay_days=delay_days
)
return {
"success": True,
"payment_intent_id": confirmed_intent.id,
"status": "authorized",
"capture_method": "manual",
"capture_scheduled_for": delay_days,
"connected_account": connected_account_id
}
def charge_with_delayed_payout(self,
customer_id: str,
amount: int,
connected_account_id: str,
delay_days: int,
payment_method_id: str):
"""For delays > 7 days: Immediate capture + delayed payout"""
# Immediate capture (money collected now)
intent = self.client.v1.payment_intents.create(
params={
"amount": amount,
"currency": "usd",
"customer": customer_id,
"payment_method": payment_method_id,
"capture_method": "automatic", # Capture immediately
"metadata": {
"delay_days": str(delay_days),
"payout_after": str(int(time.time()) + (delay_days * 24 * 60 * 60)),
"risk_review": "pending",
"delayed_payout": "true"
}
},
options={
"stripe_account": connected_account_id,
**self.req_options
}
)
# Confirm payment (money captured immediately)
confirmed_intent = self.client.v1.payment_intents.confirm(
intent.id,
options={
"stripe_account": connected_account_id,
**self.req_options
}
)
# Hold funds by disabling payouts temporarily (if needed)
# Or use your own escrow logic
self.schedule_delayed_payout_release(
payment_intent_id=confirmed_intent.id,
connected_account_id=connected_account_id,
delay_days=delay_days
)
return {
"success": True,
"payment_intent_id": confirmed_intent.id,
"status": "succeeded",
"capture_method": "immediate",
"payout_delayed_for": delay_days,
"connected_account": connected_account_id
}
def schedule_delayed_capture(self, payment_intent_id: str, connected_account_id: str, delay_days: int):
"""Schedule capture after risk review period"""
# Store in your database for processing
capture_time = datetime.now() + timedelta(days=delay_days)
self.db.create_scheduled_task({
"task_type": "capture_payment",
"payment_intent_id": payment_intent_id,
"connected_account_id": connected_account_id,
"scheduled_for": capture_time,
"status": "pending"
})
# Or use your task queue (Celery, Cloud Tasks, etc.)
# schedule_task("capture_payment", delay=delay_days*24*3600,
# args=[payment_intent_id, connected_account_id])
def schedule_delayed_payout_release(self, payment_intent_id: str, connected_account_id: str, delay_days: int):
"""Schedule payout release after risk review period"""
release_time = datetime.now() + timedelta(days=delay_days)
self.db.create_scheduled_task({
"task_type": "release_payout",
"payment_intent_id": payment_intent_id,
"connected_account_id": connected_account_id,
"scheduled_for": release_time,
"status": "pending"
})
def execute_delayed_capture(self, payment_intent_id: str, connected_account_id: str):
"""Execute scheduled capture after risk review"""
try:
# Capture the authorized payment
captured_intent = self.client.v1.payment_intents.capture(
payment_intent_id,
options={
"stripe_account": connected_account_id,
**self.req_options
}
)
return {
"success": True,
"payment_intent_id": captured_intent.id,
"status": "captured",
"message": "Payment captured after risk review period"
}
except Exception as e:
# Handle expired authorization
if "authorization" in str(e).lower():
return {
"success": False,
"error": "Authorization expired",
"payment_intent_id": payment_intent_id,
"requires_new_payment": True
}
return {"success": False, "error": str(e)}
def cancel_delayed_payment(self, payment_intent_id: str, connected_account_id: str, reason: str = "fraud_detected"):
"""Cancel payment if fraud detected during review period"""
try:
# Cancel the payment intent
cancelled_intent = self.client.v1.payment_intents.cancel(
payment_intent_id,
params={"cancellation_reason": reason},
options={
"stripe_account": connected_account_id,
**self.req_options
}
)
# Update scheduled task
self.db.update_scheduled_task(payment_intent_id, {"status": "cancelled"})
return {
"success": True,
"payment_intent_id": cancelled_intent.id,
"status": "cancelled",
"reason": reason
}
except Exception as e:
return {"success": False, "error": str(e)}
def charge_with_manual_capture(self,
customer_id: str,
amount: int,
connected_account_id: str,
delay_days: int,
payment_method_id: str):
"""For delays ≤ 7 days: Authorize now, capture later"""
# Create PaymentIntent with manual capture
intent = self.client.v1.payment_intents.create(
params={
"amount": amount,
"currency": "usd",
"customer": customer_id,
"payment_method": payment_method_id,
"capture_method": "manual", # Don't capture automatically
"metadata": {
"delay_days": str(delay_days),
"capture_after": str(int(time.time()) + (delay_days * 24 * 60 * 60))
}
},
options={"stripe_account": connected_account_id}
)
# Confirm = AUTHORIZE (hold funds, don't capture yet)
confirmed_intent = self.client.v1.payment_intents.confirm(
intent.id,
options={"stripe_account": connected_account_id}
)
# Status: requires_capture
# Schedule capture for later (after delay period)
self.schedule_delayed_capture(
payment_intent_id=confirmed_intent.id,
connected_account_id=connected_account_id,
delay_days=delay_days
)
return {
"success": True,
"payment_intent_id": confirmed_intent.id,
"status": confirmed_intent.status, # "requires_capture"
"authorization_expires_in_days": 7, # Or up to 31 with extended auth
"capture_scheduled_in_days": delay_days
}
def execute_delayed_capture(self, payment_intent_id: str, connected_account_id: str):
"""Actually capture the authorized payment after delay period"""
try:
# NOW capture the funds (this actually charges the customer)
captured_intent = self.client.v1.payment_intents.capture(
payment_intent_id,
options={"stripe_account": connected_account_id}
)
# Status changes from "requires_capture" to "succeeded"
return {
"success": True,
"payment_intent_id": captured_intent.id,
"status": captured_intent.status, # "succeeded"
"amount_captured": captured_intent.amount_captured
}
except Exception as e:
return {"success": False, "error": str(e)}
def charge_platform_with_delayed_transfer(self,
customer_id: str,
amount: int,
connected_account_id: str,
delay_days: int,
payment_method_id: str):
"""Charge platform account immediately, transfer to connected account later"""
# Create customer on PLATFORM account (not connected account)
platform_customer_id = self.get_or_create_platform_customer(customer_id)
# Charge platform account immediately
platform_intent = self.client.v1.payment_intents.create(
params={
"amount": amount,
"currency": "usd",
"customer": platform_customer_id, # Platform customer
"payment_method": payment_method_id,
"capture_method": "automatic", # Capture immediately
"metadata": {
"connected_account_id": connected_account_id,
"transfer_after": str(int(time.time()) + (delay_days * 24 * 60 * 60)),
"transfer_amount": str(amount - platform_fee)
}
}
# No stripe_account header = platform charge
)
confirmed_intent = self.client.v1.payment_intents.confirm(platform_intent.id)
# Schedule transfer to connected account after delay
self.schedule_delayed_transfer_to_connected_account(
platform_payment_intent_id=confirmed_intent.id,
connected_account_id=connected_account_id,
transfer_amount=amount - platform_fee,
delay_days=delay_days
)
return {
"success": True,
"platform_payment_intent_id": confirmed_intent.id,
"status": "captured_on_platform",
"transfer_scheduled_in_days": delay_days,
"method": "platform_escrow"
}
def execute_delayed_transfer(self,
connected_account_id: str,
amount: int):
"""Transfer funds from platform to connected account after delay"""
transfer = self.client.v1.transfers.create(
params={
"amount": amount,
"currency": "usd",
"destination": connected_account_id,
"metadata": {"delayed_transfer": "true"}
}
)
return transfer
