stripe-integration

def create_checkout_session(amount, order_attempt_id, currency='usd'):
    """Create a one-time payment checkout session."""
    try:
        session = stripe.checkout.Session.create(
            payment_method_types=['card'],
            line_items=[{
                'price_data': {
                    'currency': currency,
                    'product_data': {
                        'name': 'Purchase',
                        'images': ['https://example.com/product.jpg'],
                    },
                    'unit_amount': amount,  # Amount in cents
                },
                'quantity': 1,
            }],
            mode='payment',
            success_url='https://yourdomain.com/success?session_id={CHECKOUT_SESSION_ID}',
            cancel_url='https://yourdomain.com/cancel',
            metadata={
                'order_id': 'order_123',
                'user_id': 'user_456'
            },
            idempotency_key=order_attempt_id
        )
        return session
    except stripe.error.StripeError as e:
        # Handle error
        print(f"Stripe error: {e.user_message}")
        raise

def create_payment_intent(amount, order_attempt_id, currency='usd', customer_id=None):
    """Create a payment intent for custom checkout UI."""
    intent = stripe.PaymentIntent.create(
        amount=amount,
        currency=currency,
        customer=customer_id,
        automatic_payment_methods={
            'enabled': True,
        },
        metadata={
            'integration_check': 'accept_a_payment'
        },
        idempotency_key=order_attempt_id
    )
    return intent.client_secret  # Only to the authenticated client for this order; never log it
 
# Frontend (JavaScript)
"""
const stripe = Stripe('pk_test_...');
const elements = stripe.elements();
const cardElement = elements.create('card');
cardElement.mount('#card-element');
 
const {error, paymentIntent} = await stripe.confirmCardPayment(
    clientSecret,
    {
        payment_method: {
            card: cardElement,
            billing_details: {
                name: 'Customer Name'
            }
        }
    }
);
 
if (error) {
    // Handle error
} else if (paymentIntent.status === 'succeeded') {
    // Update display only; server fulfillment still verifies payment state
}
"""

def create_customer_portal_session(customer_id):
    """Create a portal session for customers to manage subscriptions."""
    session = stripe.billing_portal.Session.create(
        customer=customer_id,
        return_url='https://yourdomain.com/account',
    )
    return session.url  # Redirect customer here

import os
from flask import Flask, request
import stripe
 
app = Flask(__name__)
 
endpoint_secret = os.environ["STRIPE_WEBHOOK_SECRET"]
 
@app.route('/webhook', methods=['POST'])
def webhook():
    payload = request.get_data(cache=False)  # Exact raw bytes; enforce an ingress body-size limit
    sig_header = request.headers.get('Stripe-Signature')
    if not sig_header:
        return 'Missing signature', 400
 
    try:
        event = stripe.Webhook.construct_event(
            payload, sig_header, endpoint_secret
        )
    except ValueError:
        # Invalid payload
        return 'Invalid payload', 400
    except stripe.SignatureVerificationError:
        # Invalid signature
        return 'Invalid signature', 400
 
    # Integration sketch: durable deduplication/queueing below must precede effects.
    # Bind account, mode, order, amount/currency and expected current state first.
    # Handle the event
    if event['type'] == 'payment_intent.succeeded':
        payment_intent = event['data']['object']
        handle_successful_payment(payment_intent)
    elif event['type'] == 'payment_intent.payment_failed':
        payment_intent = event['data']['object']
        handle_failed_payment(payment_intent)
    elif event['type'] == 'customer.subscription.deleted':
        subscription = event['data']['object']
        handle_subscription_canceled(subscription)
 
    return 'Success', 200
 
def handle_successful_payment(payment_intent):
    """Process successful payment."""
    customer_id = payment_intent.get('customer')
    amount = payment_intent['amount']
    metadata = payment_intent.get('metadata', {})
 
    # Update your database
    # Send confirmation email
    # Fulfill order
    print(f"Payment succeeded: {payment_intent['id']}")
 
def handle_failed_payment(payment_intent):
    """Handle failed payment."""
    error = payment_intent.get('last_payment_error', {})
    print(f"Payment failed: {error.get('message')}")
    # Notify customer
    # Update order status
 
def handle_subscription_canceled(subscription):
    """Handle subscription cancellation."""
    customer_id = subscription['customer']
    # Update user access
    # Send cancellation email
    print(f"Subscription canceled: {subscription['id']}")

Verify signature and event/account/mode before accepting the event.
Atomically insert event.id into a durable inbox with a unique constraint.
Return 2xx only after durable acceptance; retryable storage failure remains a failure.
A worker reconciles current payment/order state and applies a guarded transition.
Use a unique order/fulfillment key as well: different events can describe the same payment.
Commit the state change and an outbox entry together; downstream effects are idempotent.
Duplicate/concurrent delivery returns the recorded result without fulfilling twice.

def create_customer(email, name, payment_method_id=None):
    """Create a Stripe customer."""
    customer = stripe.Customer.create(
        email=email,
        name=name,
        payment_method=payment_method_id,
        invoice_settings={
            'default_payment_method': payment_method_id
        } if payment_method_id else None,
        metadata={
            'user_id': '12345'
        }
    )
    return customer
 
def attach_payment_method(customer_id, payment_method_id):
    """Attach a payment method to a customer."""
    stripe.PaymentMethod.attach(
        payment_method_id,
        customer=customer_id
    )
 
    # Set as default
    stripe.Customer.modify(
        customer_id,
        invoice_settings={
            'default_payment_method': payment_method_id
        }
    )
 
def list_customer_payment_methods(customer_id):
    """List all payment methods for a customer."""
    payment_methods = stripe.PaymentMethod.list(
        customer=customer_id,
        type='card'
    )
    return payment_methods.data

def create_refund(payment_intent_id, refund_attempt_id, amount=None, reason=None):
    """Create a refund."""
    refund_params = {
        'payment_intent': payment_intent_id
    }
 
    if amount is not None:
        if type(amount) is not int or amount <= 0:
            raise ValueError("Refund amount must be a positive minor-unit integer")
        refund_params['amount'] = amount  # Partial refund
 
    if reason:
        refund_params['reason'] = reason  # 'duplicate', 'fraudulent', 'requested_by_customer'
 
    refund = stripe.Refund.create(**refund_params, idempotency_key=refund_attempt_id)
    return refund
 
def handle_dispute(dispute_id, evidence):
    """Update dispute with evidence."""
    stripe.Dispute.modify(
        dispute_id,
        evidence={
            'customer_name': evidence.get('customer_name'),
            'customer_email_address': evidence.get('customer_email'),
            'shipping_documentation': evidence.get('shipping_proof'),
            'customer_communication': evidence.get('communication'),
        }
    )

# Use test mode keys
import os
 
stripe.api_key = os.environ["STRIPE_SECRET_KEY"]
 
# Test card numbers
TEST_CARDS = {
    'success': '4242424242424242',
    'declined': '4000000000000002',
    '3d_secure': '4000002500003155',
    'insufficient_funds': '4000000000009995'
}
 
def test_payment_flow():
    """Test complete payment flow."""
    # Create test customer
    customer = stripe.Customer.create(
        email="test@example.com"
    )
 
    # Create payment intent
    intent = stripe.PaymentIntent.create(
        amount=1000,
        currency='usd',
        customer=customer.id,
        payment_method_types=['card']
    )
 
    # Confirm with test card
    confirmed = stripe.PaymentIntent.confirm(
        intent.id,
        payment_method='pm_card_visa'  # Test payment method
    )
 
    assert confirmed.status == 'succeeded'