import stripe from django.conf import settings import logging logger = logging.getLogger(__name__) class StripeService: """Service class to manage Stripe operations with mode awareness""" def __init__(self): # Determine if we're in test mode self.mode = getattr(settings, 'STRIPE_MODE', 'test') self.is_test_mode = self.mode == 'test' # Get appropriate keys based on mode self.api_key = settings.STRIPE_SECRET_KEY self.publishable_key = settings.STRIPE_PUBLISHABLE_KEY self.webhook_secret = settings.SHOP_STRIPE_WEBHOOK_SECRET self.currency = getattr(settings, 'STRIPE_CURRENCY', 'eur') # Configure Stripe library stripe.api_key = self.api_key # Log initialization in debug mode mode_str = "TEST" if self.is_test_mode else "LIVE" logger.debug(f"Initialized StripeService in {mode_str} mode") def create_checkout_session(self, customer_email=None, line_items=None, success_url=None, cancel_url=None, metadata=None, discounts=None): """Create a Stripe checkout session with the given parameters""" # Initialize session parameters session_params = { 'payment_method_types': ['card'], 'line_items': line_items, 'mode': 'payment', 'success_url': success_url, 'cancel_url': cancel_url, 'metadata': metadata or {}, } # Add customer email if provided if customer_email: session_params['customer_email'] = customer_email # Add discounts if provided if discounts: session_params['discounts'] = discounts # Create and return the session return stripe.checkout.Session.create(**session_params) def verify_webhook_signature(self, payload, signature): """Verify webhook signature using mode-appropriate secret""" try: event = stripe.Webhook.construct_event( payload, signature, self.webhook_secret ) return event except stripe.error.SignatureVerificationError as e: logger.error(f"Webhook signature verification failed: {str(e)}") raise def construct_event_for_testing(self, payload, signature): """Helper method to construct events during testing""" return stripe.Webhook.construct_event( payload, signature, self.webhook_secret ) def get_checkout_session(self, session_id): """Retrieve a checkout session by ID""" return stripe.checkout.Session.retrieve(session_id) def get_payment_intent(self, payment_intent_id): """Retrieve a payment intent by ID""" return stripe.PaymentIntent.retrieve(payment_intent_id) def create_refund(self, payment_intent_id, amount=None, reason=None): """ Create a refund for a payment intent Args: payment_intent_id (str): The payment intent ID to refund amount (int, optional): Amount to refund in cents. If None, refunds the entire amount. reason (str, optional): The reason for the refund, one of 'duplicate', 'fraudulent', or 'requested_by_customer' Returns: stripe.Refund: The created refund object """ try: refund_params = { 'payment_intent': payment_intent_id, } if amount is not None: refund_params['amount'] = amount if reason in ['duplicate', 'fraudulent', 'requested_by_customer']: refund_params['reason'] = reason # Log the refund attempt mode_str = "TEST" if self.is_test_mode else "LIVE" logger.info(f"[{mode_str}] Creating refund for payment intent {payment_intent_id}") # Process the refund refund = stripe.Refund.create(**refund_params) # Log success logger.info(f"Refund created successfully: {refund.id}") return refund except stripe.error.StripeError as e: # Log the error logger.error(f"Stripe error creating refund: {str(e)}") raise except Exception as e: # Log any other errors logger.error(f"Unexpected error creating refund: {str(e)}") raise def get_refund(self, refund_id): """ Retrieve a refund by ID Args: refund_id (str): The ID of the refund to retrieve Returns: stripe.Refund: The refund object """ try: return stripe.Refund.retrieve(refund_id) except stripe.error.StripeError as e: logger.error(f"Stripe error retrieving refund {refund_id}: {str(e)}") raise # Create a singleton instance for import and use throughout the app stripe_service = StripeService()