""" Atomic Stripe Client with proper 3DS/3DS2 support Implements SetupIntent-first architecture for secure payment flows Implements PaymentProvider interface for easy SDK swapping """ import stripe import uuid import logging from typing import Dict, Any, Optional from datetime import datetime, timezone from shared.config.base import BaseServiceSettings from shared.clients.payment_provider import PaymentProvider from shared.exceptions.payment_exceptions import ( PaymentVerificationError, SubscriptionCreationFailed, SetupIntentError, SubscriptionUpdateFailed, PaymentMethodError, CustomerUpdateFailed ) # Configure logging logger = logging.getLogger(__name__) class StripeClient(PaymentProvider): """ Atomic Stripe operations with proper 3DS/3DS2 support """ def __init__(self): """Initialize Stripe client with configuration""" settings = BaseServiceSettings() stripe.api_key = settings.STRIPE_SECRET_KEY # Let the SDK use its default pinned API version (2025-12-15.clover for v14.1.0) # Only override if explicitly set in environment if settings.STRIPE_API_VERSION: stripe.api_version = settings.STRIPE_API_VERSION async def create_setup_intent_for_verification( self, customer_id: str, payment_method_id: str, metadata: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: """ Create standalone SetupIntent for payment verification during registration. This is the ONLY step that happens before 3DS verification completes. NO subscription is created here - subscription is created AFTER verification. Flow: 1. Frontend collects payment method 2. Backend creates customer + SetupIntent (this method) 3. Frontend confirms SetupIntent (handles 3DS if needed) 4. Backend creates subscription AFTER SetupIntent succeeds Args: customer_id: Stripe customer ID payment_method_id: Payment method ID to verify metadata: Additional metadata for tracking Returns: SetupIntent result for frontend confirmation Raises: SetupIntentError: If SetupIntent creation fails """ try: # First attach payment method to customer try: stripe.PaymentMethod.attach( payment_method_id, customer=customer_id ) logger.info( "Payment method attached to customer", extra={"payment_method_id": payment_method_id, "customer_id": customer_id} ) except stripe.error.InvalidRequestError as e: # Payment method might already be attached if "already been attached" not in str(e): raise logger.info( "Payment method already attached to customer", extra={"payment_method_id": payment_method_id, "customer_id": customer_id} ) # Set as default payment method on customer stripe.Customer.modify( customer_id, invoice_settings={'default_payment_method': payment_method_id} ) # Create SetupIntent for verification setup_intent_params = { 'customer': customer_id, 'payment_method': payment_method_id, 'usage': 'off_session', # For future recurring payments 'confirm': True, # Confirm immediately - this triggers 3DS check 'idempotency_key': f"setup_intent_reg_{uuid.uuid4()}", 'metadata': metadata or { 'purpose': 'registration_payment_verification', 'timestamp': datetime.now(timezone.utc).isoformat() }, 'automatic_payment_methods': { 'enabled': True, 'allow_redirects': 'never' } } setup_intent = stripe.SetupIntent.create(**setup_intent_params) logger.info( "SetupIntent created for verification", extra={ "setup_intent_id": setup_intent.id, "status": setup_intent.status, "customer_id": customer_id, "payment_method_id": payment_method_id } ) # Check if 3DS is required requires_action = setup_intent.status in ['requires_action', 'requires_confirmation'] return { 'setup_intent_id': setup_intent.id, 'client_secret': setup_intent.client_secret, 'status': setup_intent.status, 'requires_action': requires_action, 'customer_id': customer_id, 'payment_method_id': payment_method_id, 'created': setup_intent.created, 'metadata': dict(setup_intent.metadata) if setup_intent.metadata else {} } except stripe.error.StripeError as e: logger.error( "SetupIntent creation for verification failed", extra={ "error": str(e), "error_type": type(e).__name__, "customer_id": customer_id, "payment_method_id": payment_method_id }, exc_info=True ) raise SetupIntentError(f"SetupIntent creation failed: {str(e)}") from e except Exception as e: logger.error( "Unexpected error creating SetupIntent for verification", extra={ "error": str(e), "customer_id": customer_id, "payment_method_id": payment_method_id }, exc_info=True ) raise SetupIntentError(f"Unexpected SetupIntent error: {str(e)}") from e # Alias for backward compatibility async def create_setup_intent_for_registration( self, customer_id: str, payment_method_id: str, metadata: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: """ Create standalone SetupIntent for payment verification during registration. This is an alias for create_setup_intent_for_verification for backward compatibility. Args: customer_id: Stripe customer ID payment_method_id: Payment method ID to verify metadata: Additional metadata for tracking Returns: SetupIntent result for frontend confirmation Raises: SetupIntentError: If SetupIntent creation fails """ return await self.create_setup_intent_for_verification( customer_id, payment_method_id, metadata ) async def create_subscription_after_verification( self, customer_id: str, price_id: str, payment_method_id: str, trial_period_days: Optional[int] = None, metadata: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: """ Create subscription AFTER SetupIntent verification succeeds. This is the SECOND step - only called after frontend confirms SetupIntent. The payment method is already verified at this point. STRIPE BEST PRACTICES FOR TRIALS: - For trial subscriptions: attach payment method to CUSTOMER (not subscription) - Use off_session=True for future merchant-initiated charges - Trial subscriptions generate $0 invoices initially - Payment method is charged automatically when trial ends Args: customer_id: Stripe customer ID price_id: Stripe price ID for the plan payment_method_id: Verified payment method ID trial_period_days: Optional trial period in days metadata: Additional metadata Returns: Subscription creation result Raises: SubscriptionCreationFailed: If subscription creation fails """ try: has_trial = trial_period_days and trial_period_days > 0 # Build base metadata base_metadata = metadata or {} base_metadata.update({ 'purpose': 'registration_subscription', 'created_after_verification': 'true', 'timestamp': datetime.now(timezone.utc).isoformat() }) # STRIPE BEST PRACTICE: For trial subscriptions, attach payment method # to CUSTOMER (not subscription) to avoid immediate charges if has_trial: # Set payment method as customer's default (already done in SetupIntent, # but ensure it's set for subscription billing) stripe.Customer.modify( customer_id, invoice_settings={'default_payment_method': payment_method_id} ) subscription_params = { 'customer': customer_id, 'items': [{'price': price_id}], 'trial_period_days': trial_period_days, 'off_session': True, # Future charges are merchant-initiated 'idempotency_key': f"sub_trial_{uuid.uuid4()}", 'payment_settings': { 'payment_method_options': { 'card': { 'request_three_d_secure': 'automatic' } }, 'save_default_payment_method': 'on_subscription' }, 'metadata': { **base_metadata, 'trial_subscription': 'true', 'trial_period_days': str(trial_period_days), 'payment_strategy': 'customer_default_method' } } logger.info( "Creating TRIAL subscription (payment method on customer)", extra={ "customer_id": customer_id, "price_id": price_id, "trial_period_days": trial_period_days } ) else: # Non-trial: attach payment method directly to subscription subscription_params = { 'customer': customer_id, 'items': [{'price': price_id}], 'default_payment_method': payment_method_id, 'idempotency_key': f"sub_immediate_{uuid.uuid4()}", 'payment_settings': { 'payment_method_options': { 'card': { 'request_three_d_secure': 'automatic' } }, 'save_default_payment_method': 'on_subscription' }, 'metadata': { **base_metadata, 'trial_subscription': 'false', 'payment_strategy': 'subscription_default_method' } } logger.info( "Creating NON-TRIAL subscription (payment method on subscription)", extra={ "customer_id": customer_id, "price_id": price_id } ) # Create subscription subscription = stripe.Subscription.create(**subscription_params) # Extract timestamps current_period_start = self._extract_timestamp( getattr(subscription, 'current_period_start', None) ) current_period_end = self._extract_timestamp( getattr(subscription, 'current_period_end', None) ) # Verify trial was set correctly for trial subscriptions if has_trial: if subscription.status != 'trialing': logger.warning( "Trial subscription created but status is not 'trialing'", extra={ "subscription_id": subscription.id, "status": subscription.status, "trial_period_days": trial_period_days, "trial_end": getattr(subscription, 'trial_end', None) } ) else: logger.info( "Trial subscription created successfully with $0 initial invoice", extra={ "subscription_id": subscription.id, "status": subscription.status, "trial_period_days": trial_period_days, "trial_end": getattr(subscription, 'trial_end', None) } ) else: logger.info( "Subscription created successfully", extra={ "subscription_id": subscription.id, "customer_id": customer_id, "status": subscription.status } ) return { 'subscription_id': subscription.id, 'customer_id': customer_id, 'status': subscription.status, 'current_period_start': current_period_start, 'current_period_end': current_period_end, 'trial_period_days': trial_period_days, 'trial_end': getattr(subscription, 'trial_end', None), 'created': getattr(subscription, 'created', None), 'metadata': dict(subscription.metadata) if subscription.metadata else {} } except stripe.error.StripeError as e: logger.error( "Subscription creation after verification failed", extra={ "error": str(e), "error_type": type(e).__name__, "customer_id": customer_id, "price_id": price_id }, exc_info=True ) raise SubscriptionCreationFailed(f"Subscription creation failed: {str(e)}") from e except Exception as e: logger.error( "Unexpected error creating subscription after verification", extra={ "error": str(e), "customer_id": customer_id, "price_id": price_id }, exc_info=True ) raise SubscriptionCreationFailed(f"Unexpected subscription error: {str(e)}") from e async def verify_setup_intent_status( self, setup_intent_id: str ) -> Dict[str, Any]: """ Atomic: Verify SetupIntent status after frontend confirmation Args: setup_intent_id: SetupIntent ID to verify Returns: SetupIntent verification result Raises: SetupIntentError: If verification fails """ try: # Retrieve the SetupIntent to check its status setup_intent = stripe.SetupIntent.retrieve(setup_intent_id) logger.info( "SetupIntent verification result", extra={ "setup_intent_id": setup_intent.id, "status": setup_intent.status, "customer_id": setup_intent.customer } ) # Check if SetupIntent was successfully verified if setup_intent.status == 'succeeded': return { 'setup_intent_id': setup_intent.id, 'status': setup_intent.status, 'customer_id': setup_intent.customer, 'payment_method_id': setup_intent.payment_method, 'verified': True, 'requires_action': False, 'last_setup_error': setup_intent.last_setup_error, 'metadata': dict(setup_intent.metadata) if setup_intent.metadata else {} } elif setup_intent.status == 'requires_action': return { 'setup_intent_id': setup_intent.id, 'status': setup_intent.status, 'customer_id': setup_intent.customer, 'payment_method_id': setup_intent.payment_method, 'verified': False, 'requires_action': True, 'client_secret': setup_intent.client_secret, 'next_action': setup_intent.next_action } else: # Failed or other status return { 'setup_intent_id': setup_intent.id, 'status': setup_intent.status, 'customer_id': setup_intent.customer, 'payment_method_id': setup_intent.payment_method, 'verified': False, 'requires_action': False, 'last_setup_error': setup_intent.last_setup_error } except stripe.error.StripeError as e: logger.error( "Stripe SetupIntent verification failed", extra={ "error": str(e), "error_type": type(e).__name__, "setup_intent_id": setup_intent_id }, exc_info=True ) raise SetupIntentError(f"SetupIntent verification failed: {str(e)}") from e except Exception as e: logger.error( "Unexpected error verifying SetupIntent", extra={ "error": str(e), "setup_intent_id": setup_intent_id }, exc_info=True ) raise SetupIntentError(f"Unexpected verification error: {str(e)}") from e async def create_subscription_with_verified_payment( self, customer_id: str, price_id: str, payment_method_id: str, trial_period_days: Optional[int] = None, billing_cycle_anchor: Optional[Any] = None, metadata: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: """ Atomic: Create subscription with verified payment method ONLY called after successful SetupIntent verification Args: customer_id: Stripe customer ID price_id: Stripe price ID payment_method_id: Verified payment method ID trial_period_days: Optional trial period in days billing_cycle_anchor: Optional billing cycle anchor - can be: - int: Unix timestamp for future billing anchor - str "now": Start billing immediately (default behavior) - str "unchanged": Keep existing anchor (for plan changes) metadata: Additional metadata Returns: Subscription creation result with keys: - subscription_id: Stripe subscription ID - customer_id: Stripe customer ID - status: Subscription status - current_period_start: ISO timestamp - current_period_end: ISO timestamp - payment_intent_id: Payment intent ID if applicable - requires_action: Boolean if payment action needed - client_secret: Client secret for payment verification - trial_period_days: Trial period days if applicable - billing_cycle_anchor: Billing cycle anchor - created: Subscription creation timestamp - metadata: Subscription metadata Raises: SubscriptionCreationFailed: If subscription creation fails Stripe Best Practices: - For trials: Attach payment method to customer, not subscription - For non-trials: Attach payment method to subscription - Always use 3D Secure for card payments - Use proper billing_cycle_anchor for trial timing - Handle requires_action for payment verification """ try: # Build base subscription parameters subscription_params = { 'customer': customer_id, 'items': [{'price': price_id}], 'expand': ['latest_invoice.payment_intent'], 'idempotency_key': f"subscription_{uuid.uuid4()}", 'proration_behavior': 'none', # Default to no proration 'payment_settings': { 'payment_method_options': { 'card': { 'request_three_d_secure': 'automatic' # Always use 3DS } }, 'save_default_payment_method': 'on_subscription' } } # Handle metadata - preserve original and add base metadata base_metadata = metadata or { 'purpose': 'registration_subscription', 'timestamp': datetime.now().isoformat() } # Add trial-specific parameters if trial period is specified if trial_period_days is not None and trial_period_days > 0: subscription_params['trial_period_days'] = trial_period_days # Note: trial_from_plan is NOT used here because it conflicts with trial_period_days # Stripe API: "You cannot set trial_end or trial_period_days when trial_from_plan=true" subscription_params['off_session'] = True # For trial subscriptions, don't set billing_cycle_anchor # Stripe will automatically handle the trial timing # Remove billing_cycle_anchor if it was set for non-trial subscriptions subscription_params.pop('billing_cycle_anchor', None) # Add trial-specific metadata base_metadata.update({ 'trial_subscription': 'true', 'trial_period_days': str(trial_period_days), 'no_immediate_payment': 'true', 'trial_start_date': datetime.now().isoformat(), 'payment_strategy': 'attach_to_customer' }) # For trial subscriptions, attach payment method to customer FIRST # This is required by Stripe for subscriptions with trial periods try: stripe.PaymentMethod.attach( payment_method_id, customer=customer_id ) logger.info("Payment method attached to customer for trial subscription", extra={"payment_method_id": payment_method_id, "customer_id": customer_id}) except Exception as e: logger.error("Failed to attach payment method to customer for trial subscription", extra={"error": str(e), "payment_method_id": payment_method_id, "customer_id": customer_id}, exc_info=True) raise SubscriptionCreationFailed(f"Failed to attach payment method: {str(e)}") from e # THEN set it as default payment method on customer # This prevents immediate charge while ensuring payment method is available stripe.Customer.modify( customer_id, invoice_settings={ 'default_payment_method': payment_method_id } ) else: # For non-trial subscriptions, attach payment method to subscription subscription_params['default_payment_method'] = payment_method_id subscription_params['proration_behavior'] = 'create_prorations' # Add non-trial metadata base_metadata.update({ 'payment_strategy': 'attach_to_subscription', 'trial_period_days': '0' }) # Handle billing cycle anchor parameter (override if explicitly provided) # But for trial subscriptions, always use "now" to start trial immediately if trial_period_days is not None and trial_period_days > 0: # Trial subscription - billing_cycle_anchor was removed earlier pass # Don't set billing_cycle_anchor for trials elif billing_cycle_anchor is not None: # Stripe requires billing_cycle_anchor to be a Unix timestamp (integer) # If "now" is passed, don't set it - Stripe defaults to immediate start # If "unchanged" is passed, also skip it (used for plan changes) if isinstance(billing_cycle_anchor, str): if billing_cycle_anchor.lower() in ('now', 'unchanged'): # Skip setting billing_cycle_anchor - Stripe will use current time pass else: # Try to parse as integer timestamp try: subscription_params['billing_cycle_anchor'] = int(billing_cycle_anchor) except ValueError: logger.warning( "Invalid billing_cycle_anchor value, skipping", extra={"billing_cycle_anchor": billing_cycle_anchor} ) elif isinstance(billing_cycle_anchor, int): subscription_params['billing_cycle_anchor'] = billing_cycle_anchor subscription_params['metadata'] = base_metadata # Log subscription parameters for debugging (redact sensitive data) logger.debug( "Creating Stripe subscription with parameters", extra={ "customer_id": customer_id, "price_id": price_id, "trial_period_days": trial_period_days, "billing_cycle_anchor": billing_cycle_anchor, "payment_method_id": payment_method_id, "metadata_keys": list(base_metadata.keys()) if base_metadata else [] } ) # Create subscription try: stripe_subscription = stripe.Subscription.create(**subscription_params) except Exception as e: logger.error( "Stripe API call failed during subscription creation", extra={ "error": str(e), "customer_id": customer_id, "price_id": price_id, "trial_period_days": trial_period_days, "billing_cycle_anchor": billing_cycle_anchor }, exc_info=True ) raise SubscriptionCreationFailed(f"Stripe API failed: {str(e)}") from e # Validate that we got a proper subscription object if not hasattr(stripe_subscription, 'id') or not hasattr(stripe_subscription, 'status'): logger.error( "Invalid Stripe API response - not a subscription object", extra={ "response_type": type(stripe_subscription).__name__, "response_keys": list(stripe_subscription.keys()) if hasattr(stripe_subscription, 'keys') else 'N/A', "customer_id": customer_id, "price_id": price_id } ) raise SubscriptionCreationFailed( f"Invalid Stripe API response: expected Subscription object, got {type(stripe_subscription).__name__}" ) # Validate subscription creation status = stripe_subscription.status # Check if subscription is in a valid state before extracting timestamps if status not in ['trialing', 'active', 'incomplete', 'past_due', 'canceled', 'unpaid']: subscription_id = getattr(stripe_subscription, 'id', 'unknown') raise SubscriptionCreationFailed( f"Invalid subscription status: {status}. " f"Subscription ID: {subscription_id}" ) # For incomplete subscriptions, we can't extract timestamps as they may not be set yet if status == 'incomplete': subscription_id = getattr(stripe_subscription, 'id', 'unknown') logger.warning( "Incomplete subscription created - timestamps may not be available", extra={ "subscription_id": subscription_id, "customer_id": customer_id, "status": status } ) # Set timestamps to None for incomplete subscriptions current_period_start = None current_period_end = None else: # Extract period dates using helper method for valid subscriptions logger.debug( "Extracting timestamps from Stripe subscription", extra={ "current_period_start_raw": getattr(stripe_subscription, 'current_period_start', None), "current_period_start_type": type(getattr(stripe_subscription, 'current_period_start', None)).__name__, "current_period_end_raw": getattr(stripe_subscription, 'current_period_end', None), "current_period_end_type": type(getattr(stripe_subscription, 'current_period_end', None)).__name__ } ) current_period_start = self._extract_timestamp(getattr(stripe_subscription, 'current_period_start', None)) current_period_end = self._extract_timestamp(getattr(stripe_subscription, 'current_period_end', None)) logger.debug( "Extracted timestamps", extra={ "current_period_start": current_period_start, "current_period_end": current_period_end, "subscription_id": getattr(stripe_subscription, 'id', 'unknown') } ) # Check payment requirements requires_action = False client_secret = None payment_intent_id = None setup_intent_id = None # First check PaymentIntent from latest_invoice (for non-trial subscriptions) if (stripe_subscription.latest_invoice and hasattr(stripe_subscription.latest_invoice, 'payment_intent')): payment_intent = stripe_subscription.latest_invoice.payment_intent if payment_intent: payment_intent_id = payment_intent.id if payment_intent.status in ['requires_action', 'requires_source_action']: requires_action = True client_secret = payment_intent.client_secret # For trial subscriptions, check pending_setup_intent for 3DS requirement # This is critical: trial subscriptions have $0 invoices with no PaymentIntent, # but Stripe creates a SetupIntent to verify the card for future payments pending_setup_intent = getattr(stripe_subscription, 'pending_setup_intent', None) if pending_setup_intent and not requires_action: # pending_setup_intent can be a string ID or an expanded object if isinstance(pending_setup_intent, str): # Need to retrieve the SetupIntent to check its status try: setup_intent = stripe.SetupIntent.retrieve(pending_setup_intent) setup_intent_id = setup_intent.id if setup_intent.status in ['requires_action', 'requires_confirmation']: requires_action = True client_secret = setup_intent.client_secret logger.info( "Trial subscription requires SetupIntent verification (3DS)", extra={ "subscription_id": stripe_subscription.id, "setup_intent_id": setup_intent_id, "setup_intent_status": setup_intent.status } ) except Exception as e: logger.warning( "Failed to retrieve pending_setup_intent, continuing without 3DS check", extra={"setup_intent_id": pending_setup_intent, "error": str(e)} ) elif hasattr(pending_setup_intent, 'id'): # Already expanded setup_intent_id = pending_setup_intent.id if pending_setup_intent.status in ['requires_action', 'requires_confirmation']: requires_action = True client_secret = pending_setup_intent.client_secret logger.info( "Trial subscription requires SetupIntent verification (3DS)", extra={ "subscription_id": stripe_subscription.id, "setup_intent_id": setup_intent_id, "setup_intent_status": pending_setup_intent.status } ) # Validate trial subscriptions if trial_period_days is not None and trial_period_days > 0: self._validate_trial_subscription( stripe_subscription, trial_period_days, requires_action ) logger.info( "Subscription created successfully", extra={ "subscription_id": stripe_subscription.id, "customer_id": customer_id, "status": status, "requires_action": requires_action, "setup_intent_id": setup_intent_id, "trial_period_days": trial_period_days } ) return { 'subscription_id': stripe_subscription.id, 'customer_id': customer_id, 'status': status, 'current_period_start': current_period_start, 'current_period_end': current_period_end, 'payment_intent_id': payment_intent_id, 'setup_intent_id': setup_intent_id, 'requires_action': requires_action, 'client_secret': client_secret, 'trial_period_days': trial_period_days, 'billing_cycle_anchor': getattr(stripe_subscription, 'billing_cycle_anchor', None), 'created': getattr(stripe_subscription, 'created', None), 'metadata': getattr(stripe_subscription, 'metadata', {}) } except stripe.error.StripeError as e: logger.error( "Stripe subscription creation failed", extra={ "error": str(e), "error_type": type(e).__name__, "customer_id": customer_id, "price_id": price_id }, exc_info=True ) raise SubscriptionCreationFailed(f"Subscription creation failed: {str(e)}") from e except Exception as e: logger.error( "Unexpected error creating subscription", extra={ "error": str(e), "customer_id": customer_id, "price_id": price_id }, exc_info=True ) raise SubscriptionCreationFailed(f"Unexpected subscription error: {str(e)}") from e def _validate_trial_subscription(self, stripe_subscription, trial_period_days, requires_action): """Validate that trial subscription was created correctly""" # Check status if stripe_subscription.status not in ['trialing', 'active']: raise SubscriptionCreationFailed( f"Invalid trial subscription status: {stripe_subscription.status}. " f"Expected 'trialing' or 'active'. Subscription ID: {stripe_subscription.id}" ) # Check trial end is set if not stripe_subscription.trial_end: raise SubscriptionCreationFailed( f"Trial subscription missing trial_end. Subscription ID: {stripe_subscription.id}" ) # Check for immediate payment requirements during trial if (stripe_subscription.latest_invoice and hasattr(stripe_subscription.latest_invoice, 'amount_due')): amount_due = stripe_subscription.latest_invoice.amount_due current_time = int(datetime.now().timestamp()) # If we're still in the trial period if stripe_subscription.trial_end > current_time: if amount_due > 0: # This is problematic - trial should have $0 due logger.warning( "Trial subscription has non-zero amount due during trial", extra={ "subscription_id": stripe_subscription.id, "amount_due": amount_due, "trial_period_days": trial_period_days, "trial_end": stripe_subscription.trial_end, "current_time": current_time } ) # For now, we allow this but log it - the frontend can handle payment verification else: # This is expected - trial with $0 due logger.info( "Trial subscription created correctly with $0 amount due", extra={ "subscription_id": stripe_subscription.id, "trial_period_days": trial_period_days, "trial_end": stripe_subscription.trial_end } ) def _extract_timestamp(self, timestamp): """Extract and format timestamp from Stripe response""" if not timestamp: return None try: if isinstance(timestamp, int): return datetime.fromtimestamp(timestamp, tz=timezone.utc).isoformat() elif hasattr(timestamp, 'isoformat'): return timestamp.isoformat() elif isinstance(timestamp, str): # Handle string timestamps - try to parse as ISO format first try: # Try to parse as ISO format parsed_dt = datetime.fromisoformat(timestamp) if parsed_dt.tzinfo is None: parsed_dt = parsed_dt.replace(tzinfo=timezone.utc) return parsed_dt.isoformat() except ValueError: # If not ISO format, try to parse as Unix timestamp try: return datetime.fromtimestamp(float(timestamp), tz=timezone.utc).isoformat() except (ValueError, TypeError): # If all parsing fails, return the string as-is return timestamp else: return str(timestamp) except Exception as e: logger.error( "Failed to extract timestamp from Stripe response", extra={ "timestamp_value": str(timestamp), "timestamp_type": type(timestamp).__name__, "error": str(e) }, exc_info=True ) # Return the raw value as string if conversion fails return str(timestamp) async def create_customer( self, email: str, name: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: """ Atomic: Create Stripe customer Args: email: Customer email name: Customer name metadata: Additional metadata Returns: Customer creation result Raises: Exception: If customer creation fails """ try: customer_params = { 'email': email, 'idempotency_key': f"customer_{uuid.uuid4()}", 'metadata': metadata or { 'purpose': 'registration_customer', 'timestamp': datetime.now().isoformat() } } if name: customer_params['name'] = name customer = stripe.Customer.create(**customer_params) logger.info( "Stripe customer created", extra={ "customer_id": customer.id, "email": email } ) return customer except stripe.error.StripeError as e: logger.error( "Stripe customer creation failed", extra={ "error": str(e), "error_type": type(e).__name__, "email": email }, exc_info=True ) raise Exception(f"Customer creation failed: {str(e)}") from e except Exception as e: logger.error( "Unexpected error creating customer", extra={ "error": str(e), "email": email }, exc_info=True ) raise Exception(f"Unexpected customer error: {str(e)}") from e async def attach_payment_method( self, payment_method_id: str, customer_id: str ) -> Dict[str, Any]: """ Atomic: Attach payment method to customer Args: payment_method_id: Payment method ID customer_id: Customer ID Returns: Payment method attachment result Raises: Exception: If attachment fails """ try: payment_method = stripe.PaymentMethod.attach( payment_method_id, customer=customer_id, idempotency_key=f"attach_pm_{uuid.uuid4()}" ) logger.info( "Payment method attached to customer", extra={ "payment_method_id": payment_method.id, "customer_id": customer_id } ) return { 'payment_method_id': payment_method.id, 'customer_id': payment_method.customer, 'card_last4': payment_method.card.last4 if payment_method.card else None, 'card_brand': payment_method.card.brand if payment_method.card else None } except stripe.error.InvalidRequestError as e: if 'already been attached' in str(e): logger.warning( "Payment method already attached", extra={ "payment_method_id": payment_method_id, "customer_id": customer_id } ) return { 'payment_method_id': payment_method_id, 'customer_id': customer_id, 'already_attached': True } else: raise Exception(f"Payment method attachment failed: {str(e)}") from e except Exception as e: logger.error( "Unexpected error attaching payment method", extra={ "error": str(e), "payment_method_id": payment_method_id, "customer_id": customer_id }, exc_info=True ) raise Exception(f"Unexpected attachment error: {str(e)}") from e async def set_default_payment_method( self, customer_id: str, payment_method_id: str ) -> Dict[str, Any]: """ Atomic: Set default payment method for customer Args: customer_id: Customer ID payment_method_id: Payment method ID Returns: Default payment method update result Raises: Exception: If update fails """ try: customer = stripe.Customer.modify( customer_id, invoice_settings={ 'default_payment_method': payment_method_id }, idempotency_key=f"default_pm_{uuid.uuid4()}" ) logger.info( "Default payment method updated", extra={ "customer_id": customer_id, "payment_method_id": payment_method_id } ) return { 'customer_id': customer.id, 'default_payment_method': customer.invoice_settings.default_payment_method } except Exception as e: logger.error( "Failed to update default payment method", extra={ "error": str(e), "customer_id": customer_id, "payment_method_id": payment_method_id }, exc_info=True ) raise Exception(f"Default payment method update failed: {str(e)}") from e async def create_setup_intent(self) -> Dict[str, Any]: """ Create a basic SetupIntent Returns: SetupIntent creation result """ try: setup_intent = stripe.SetupIntent.create( usage='off_session', idempotency_key=f"setup_intent_{uuid.uuid4()}" ) logger.info( "Basic SetupIntent created", extra={ "setup_intent_id": setup_intent.id } ) return { 'setup_intent_id': setup_intent.id, 'client_secret': setup_intent.client_secret, 'status': setup_intent.status } except stripe.error.StripeError as e: logger.error( "Basic SetupIntent creation failed", extra={ "error": str(e) }, exc_info=True ) raise SetupIntentError(f"SetupIntent creation failed: {str(e)}") from e async def get_setup_intent(self, setup_intent_id: str) -> Any: """ Get SetupIntent details Args: setup_intent_id: SetupIntent ID Returns: SetupIntent object """ try: setup_intent = stripe.SetupIntent.retrieve(setup_intent_id) return setup_intent except stripe.error.StripeError as e: logger.error( "Failed to retrieve SetupIntent", extra={ "error": str(e), "setup_intent_id": setup_intent_id }, exc_info=True ) raise SetupIntentError(f"Failed to retrieve SetupIntent: {str(e)}") from e async def create_payment_intent( self, amount: float, currency: str, customer_id: str, payment_method_id: str ) -> Dict[str, Any]: """ Create a PaymentIntent for one-time payments Args: amount: Payment amount currency: Currency code customer_id: Customer ID payment_method_id: Payment method ID Returns: PaymentIntent creation result """ try: payment_intent = stripe.PaymentIntent.create( amount=int(amount * 100), # Convert to cents currency=currency, customer=customer_id, payment_method=payment_method_id, confirm=True, idempotency_key=f"payment_intent_{uuid.uuid4()}" ) logger.info( "PaymentIntent created", extra={ "payment_intent_id": payment_intent.id, "status": payment_intent.status } ) return { 'payment_intent_id': payment_intent.id, 'status': payment_intent.status, 'client_secret': payment_intent.client_secret, 'requires_action': payment_intent.status == 'requires_action' } except stripe.error.StripeError as e: logger.error( "PaymentIntent creation failed", extra={ "error": str(e) }, exc_info=True ) raise PaymentVerificationError(f"PaymentIntent creation failed: {str(e)}") from e async def complete_subscription_after_setup_intent( self, setup_intent_id: str ) -> Dict[str, Any]: """ Complete subscription creation after SetupIntent verification Note: This retrieves the SetupIntent and returns its details. The actual subscription creation should use create_subscription_with_verified_payment. Args: setup_intent_id: SetupIntent ID Returns: SetupIntent verification result """ return await self.verify_setup_intent_status(setup_intent_id) async def cancel_subscription( self, subscription_id: str ) -> Dict[str, Any]: """ Cancel a subscription at period end Args: subscription_id: Subscription ID Returns: Subscription cancellation result """ try: subscription = stripe.Subscription.modify( subscription_id, cancel_at_period_end=True, idempotency_key=f"cancel_sub_{uuid.uuid4()}" ) logger.info( "Subscription set to cancel at period end", extra={ "subscription_id": subscription_id, "cancel_at_period_end": subscription.cancel_at_period_end } ) return { 'subscription_id': subscription.id, 'status': subscription.status, 'cancel_at_period_end': subscription.cancel_at_period_end, 'current_period_end': subscription.current_period_end } except stripe.error.StripeError as e: logger.error( "Subscription cancellation failed", extra={ "error": str(e), "subscription_id": subscription_id }, exc_info=True ) raise SubscriptionCreationFailed(f"Subscription cancellation failed: {str(e)}") from e async def update_payment_method( self, customer_id: str, payment_method_id: str ) -> Dict[str, Any]: """ Update customer's payment method Args: customer_id: Customer ID payment_method_id: New payment method ID Returns: Payment method update result """ try: # Attach payment method to customer await self.attach_payment_method(payment_method_id, customer_id) # Set as default result = await self.set_default_payment_method(customer_id, payment_method_id) return result except Exception as e: logger.error( "Failed to update payment method", extra={ "error": str(e), "customer_id": customer_id, "payment_method_id": payment_method_id }, exc_info=True ) raise Exception(f"Payment method update failed: {str(e)}") from e async def update_subscription( self, subscription_id: str, new_price_id: str ) -> Dict[str, Any]: """ Update subscription price (plan upgrade/downgrade) Args: subscription_id: Subscription ID new_price_id: New price ID Returns: Subscription update result """ try: subscription = stripe.Subscription.retrieve(subscription_id) updated_subscription = stripe.Subscription.modify( subscription_id, items=[{ 'id': subscription['items']['data'][0].id, 'price': new_price_id, }], proration_behavior='create_prorations', idempotency_key=f"update_sub_{uuid.uuid4()}" ) logger.info( "Subscription updated", extra={ "subscription_id": subscription_id, "new_price_id": new_price_id } ) return { 'subscription_id': updated_subscription.id, 'status': updated_subscription.status, 'current_period_end': updated_subscription.current_period_end } except stripe.error.StripeError as e: logger.error( "Subscription update failed", extra={ "error": str(e), "subscription_id": subscription_id }, exc_info=True ) raise SubscriptionCreationFailed(f"Subscription update failed: {str(e)}") from e async def get_subscription( self, subscription_id: str ) -> Dict[str, Any]: """ Get subscription details Args: subscription_id: Subscription ID Returns: Subscription details """ try: subscription = stripe.Subscription.retrieve(subscription_id) # Safely extract current_period_start (may not be present in all subscription types) current_period_start = None try: # Use getattr with default to safely access the attribute current_period_start = getattr(subscription, 'current_period_start', None) if current_period_start: # Convert to ISO format if it's a timestamp if isinstance(current_period_start, int): current_period_start = datetime.fromtimestamp(current_period_start, tz=timezone.utc).isoformat() elif hasattr(current_period_start, 'isoformat'): current_period_start = current_period_start.isoformat() elif isinstance(current_period_start, str): # Handle string timestamps try: # Try to parse as ISO format parsed_dt = datetime.fromisoformat(current_period_start) if parsed_dt.tzinfo is None: parsed_dt = parsed_dt.replace(tzinfo=timezone.utc) current_period_start = parsed_dt.isoformat() except ValueError: # If not ISO format, try to parse as Unix timestamp try: current_period_start = datetime.fromtimestamp(float(current_period_start), tz=timezone.utc).isoformat() except (ValueError, TypeError): # If all parsing fails, keep as string pass except Exception as e: logger.warning( "current_period_start not available in retrieved subscription", extra={ "subscription_id": subscription.id, "status": subscription.status, "error": str(e) } ) current_period_start = None # Safely extract current_period_end current_period_end = None try: # Use getattr with default to safely access the attribute current_period_end = getattr(subscription, 'current_period_end', None) if current_period_end: # Convert to ISO format if it's a timestamp if isinstance(current_period_end, int): current_period_end = datetime.fromtimestamp(current_period_end, tz=timezone.utc).isoformat() elif hasattr(current_period_end, 'isoformat'): current_period_end = current_period_end.isoformat() elif isinstance(current_period_end, str): # Handle string timestamps try: # Try to parse as ISO format parsed_dt = datetime.fromisoformat(current_period_end) if parsed_dt.tzinfo is None: parsed_dt = parsed_dt.replace(tzinfo=timezone.utc) current_period_end = parsed_dt.isoformat() except ValueError: # If not ISO format, try to parse as Unix timestamp try: current_period_end = datetime.fromtimestamp(float(current_period_end), tz=timezone.utc).isoformat() except (ValueError, TypeError): # If all parsing fails, keep as string pass except Exception as e: logger.warning( "current_period_end not available in retrieved subscription", extra={ "subscription_id": subscription.id, "status": subscription.status, "error": str(e) } ) current_period_end = None return { 'subscription_id': subscription.id, 'status': subscription.status, 'current_period_start': current_period_start, 'current_period_end': current_period_end, 'cancel_at_period_end': subscription.cancel_at_period_end, 'customer_id': subscription.customer } except stripe.error.StripeError as e: logger.error("Failed to retrieve subscription", error=str(e), subscription_id=subscription_id, exc_info=True) raise SubscriptionCreationFailed(f"Failed to retrieve subscription: {str(e)}") from e async def get_customer_payment_method( self, customer_id: str ) -> Dict[str, Any]: """ Get customer's default payment method Args: customer_id: Customer ID Returns: Payment method details """ try: customer = stripe.Customer.retrieve(customer_id) default_pm_id = None if customer.invoice_settings and customer.invoice_settings.default_payment_method: default_pm_id = customer.invoice_settings.default_payment_method if not default_pm_id: return None payment_method = stripe.PaymentMethod.retrieve(default_pm_id) return { 'payment_method_id': payment_method.id, 'type': payment_method.type, 'brand': payment_method.card.brand if payment_method.card else None, 'last4': payment_method.card.last4 if payment_method.card else None, 'exp_month': payment_method.card.exp_month if payment_method.card else None, 'exp_year': payment_method.card.exp_year if payment_method.card else None } except stripe.error.StripeError as e: logger.error("Failed to retrieve customer payment method", error=str(e), customer_id=customer_id, exc_info=True) return None async def get_invoices( self, customer_id: str ) -> Dict[str, Any]: """ Get customer invoices Args: customer_id: Customer ID Returns: List of invoices """ try: invoices = stripe.Invoice.list(customer=customer_id, limit=10) return { 'invoices': [ { 'id': inv.id, 'amount_due': inv.amount_due / 100, 'amount_paid': inv.amount_paid / 100, 'status': inv.status, 'created': inv.created, 'invoice_pdf': inv.invoice_pdf } for inv in invoices.data ] } except stripe.error.StripeError as e: logger.error("Failed to retrieve invoices", error=str(e), customer_id=customer_id, exc_info=True) return {'invoices': []} async def update_subscription_payment_method( self, subscription_id: str, payment_method_id: str ) -> Any: """ Update an existing subscription with a new payment method Args: subscription_id: Stripe subscription ID payment_method_id: New payment method ID Returns: Updated Stripe subscription object Raises: SubscriptionUpdateFailed: If the update fails """ try: logger.info("Updating subscription payment method in Stripe", subscription_id=subscription_id, payment_method_id=payment_method_id) # Update the subscription's default payment method stripe_subscription = stripe.Subscription.modify( subscription_id, default_payment_method=payment_method_id ) logger.info("Subscription payment method updated in Stripe", subscription_id=subscription_id, status=stripe_subscription.status) return stripe_subscription except Exception as e: logger.error( "Failed to update subscription payment method in Stripe", extra={ "subscription_id": subscription_id, "payment_method_id": payment_method_id, "error": str(e) }, exc_info=True ) raise SubscriptionUpdateFailed(f"Stripe API failed: {str(e)}") from e async def attach_payment_method_to_customer( self, customer_id: str, payment_method_id: str ) -> Any: """ Attach a payment method to a customer Args: customer_id: Stripe customer ID payment_method_id: Payment method ID Returns: Updated payment method object Raises: PaymentMethodError: If the attachment fails """ try: logger.info("Attaching payment method to customer in Stripe", customer_id=customer_id, payment_method_id=payment_method_id) # Attach payment method to customer payment_method = stripe.PaymentMethod.attach( payment_method_id, customer=customer_id ) logger.info("Payment method attached to customer in Stripe", customer_id=customer_id, payment_method_id=payment_method.id) return payment_method except Exception as e: logger.error( "Failed to attach payment method to customer in Stripe", extra={ "customer_id": customer_id, "payment_method_id": payment_method_id, "error": str(e) }, exc_info=True ) raise PaymentMethodError(f"Stripe API failed: {str(e)}") from e async def set_customer_default_payment_method( self, customer_id: str, payment_method_id: str ) -> Any: """ Set a payment method as the customer's default payment method Args: customer_id: Stripe customer ID payment_method_id: Payment method ID Returns: Updated customer object Raises: CustomerUpdateFailed: If the update fails """ try: logger.info("Setting default payment method for customer in Stripe", customer_id=customer_id, payment_method_id=payment_method_id) # Set default payment method on customer customer = stripe.Customer.modify( customer_id, invoice_settings={ 'default_payment_method': payment_method_id } ) logger.info("Default payment method set for customer in Stripe", customer_id=customer.id, payment_method_id=payment_method_id) return customer except Exception as e: logger.error( "Failed to set default payment method for customer in Stripe", extra={ "customer_id": customer_id, "payment_method_id": payment_method_id, "error": str(e) }, exc_info=True ) raise CustomerUpdateFailed(f"Stripe API failed: {str(e)}") from e # Singleton instance for dependency injection stripe_client = StripeClient()