stripe-integration

Stripe 統合

チェックアウト、サブスクリプション、Webhook、返金など、堅牢で PCI 準拠の決済フローを実現する Stripe 決済処理統合を習得します。

このスキルを使用する場面

• Web／モバイルアプリケーションに決済処理を実装する場合
• サブスクリプション課金システムを設定する場合
• 単発決済や継続課金を処理する場合
• 返金や異議申し立てを処理する場合
• 顧客の決済方法を管理する場合
• 欧州での決済に SCA（強力な顧客認証）を実装する場合
• Stripe Connect でマーケットプレイスの決済フローを構築する場合

中核となる概念

1. 決済フロー

Checkout Sessions

• ほとんどの統合で推奨されます
• あらゆる UI パスに対応します。
  • Stripe がホストするチェックアウトページ
  • 埋め込み型チェックアウトフォーム
  • ui_mode='custom' を使用する Elements（Payment Element、Express Checkout Element）によるカスタム UI
• チェックアウト機能（明細項目、割引、税金、配送、住所の収集、保存済み決済方法、チェックアウトのライフサイクルイベント）が組み込まれています
• Payment Intents よりも統合と保守の負担が軽減されます

Payment Intents（柔軟な制御）

• 税金、割引、サブスクリプション、通貨換算を考慮した最終金額を、自分で計算します。
• 実装がより複雑で、長期的な保守の負担も大きくなります
• PCI 準拠のために Stripe.js が必要です

Setup Intents（決済方法の保存）

• 課金せずに決済方法を収集します
• サブスクリプションや将来の決済に使用します
• 顧客による確認が必要です

2. Webhook

重要なイベント：

• payment_intent.succeeded：決済が完了しました
• payment_intent.payment_failed：決済に失敗しました
• customer.subscription.updated：サブスクリプションが変更されました
• customer.subscription.deleted：サブスクリプションがキャンセルされました
• charge.refunded：返金が処理されました
• invoice.payment_succeeded：サブスクリプションの決済に成功しました

3. サブスクリプション

構成要素：

• Product：販売するもの
• Price：金額と請求頻度
• Subscription：顧客の継続課金
• Invoice：各請求サイクルで生成される請求書

4. 顧客管理

• 顧客レコードの作成と管理
• 複数の決済方法の保存
• 顧客のメタデータの追跡
• 請求情報の管理

クイックスタート

import stripe

stripe.api_key = "sk_test_..."

# Create a checkout session
session = stripe.checkout.Session.create(
    line_items=[{
        'price_data': {
            'currency': 'usd',
            'product_data': {
                'name': 'Premium Subscription',
            },
            'unit_amount': 2000,  # $20.00
            'recurring': {
                'interval': 'month',
            },
        },
        'quantity': 1,
    }],
    mode='subscription',
    success_url='https://yourdomain.com/success?session_id={CHECKOUT_SESSION_ID}',
    cancel_url='https://yourdomain.com/cancel'
)

# Redirect user to session.url
print(session.url)

決済の実装パターン

パターン 1：単発決済（ホスト型チェックアウト）

def create_checkout_session(amount, currency='usd'):
    """Create a one-time payment checkout session."""
    try:
        session = stripe.checkout.Session.create(
            line_items=[{
                'price_data': {
                    'currency': currency,
                    'product_data': {
                        'name': 'Blue T-shirt',
                        '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'
            }
        )
        return session
    except stripe.error.StripeError as e:
        # Handle error
        print(f"Stripe error: {e.user_message}")
        raise

パターン 2：Checkout Sessions で Elements を使用

def create_checkout_session_for_elements(amount, currency='usd'):
    """Create a checkout session configured for Payment Element."""
    session = stripe.checkout.Session.create(
        mode='payment',
        ui_mode='custom',
        line_items=[{
            'price_data': {
                'currency': currency,
                'product_data': {'name': 'Blue T-shirt'},
                'unit_amount': amount,
            },
            'quantity': 1,
        }],
        return_url='https://yourdomain.com/complete?session_id={CHECKOUT_SESSION_ID}'
    )
    return session.client_secret  # Send to frontend

const stripe = Stripe("pk_test_...");
const appearance = { theme: "stripe" };

const checkout = stripe.initCheckout({
  clientSecret,
  elementsOptions: { appearance },
});
const loadActionsResult = await checkout.loadActions();

if (loadActionsResult.type === "success") {
  const { actions } = loadActionsResult;
  const session = actions.getSession();

  const button = document.getElementById("pay-button");
  const checkoutContainer = document.getElementById("checkout-container");
  const emailInput = document.getElementById("email");
  const emailErrors = document.getElementById("email-errors");
  const errors = document.getElementById("confirm-errors");

  // Display a formatted string representing the total amount
  checkoutContainer.append(`Total: ${session.total.total.amount}`);

  // Mount Payment Element
  const paymentElement = checkout.createPaymentElement();
  paymentElement.mount("#payment-element");

  // Store email for submission
  emailInput.addEventListener("blur", () => {
    actions.updateEmail(emailInput.value).then(result => {
      if (result.error) emailErrors.textContent = result.error.message;
    });
  });

  // Handle form submission
  button.addEventListener("click", () => {
    actions.confirm().then(result => {
      if (result.type === "error") errors.textContent = result.error.message;
    });
  });
}

パターン 3：Payment Intents で Elements を使用

パターン 2（Checkout Sessions で Elements を使用）が Stripe の推奨する方法ですが、代替手段として Payment Intents も使用できます。

def create_payment_intent(amount, currency='usd', customer_id=None):
    """Create a payment intent for bespoke checkout UI with Payment Element."""
    intent = stripe.PaymentIntent.create(
        amount=amount,
        currency=currency,
        customer=customer_id,
        automatic_payment_methods={
            'enabled': True,
        },
        metadata={
            'integration_check': 'accept_a_payment'
        }
    )
    return intent.client_secret  # Send to frontend

// Mount Payment Element and confirm via Payment Intents
const stripe = Stripe("pk_test_...");
const appearance = { theme: "stripe" };
const elements = stripe.elements({ appearance, clientSecret });

const paymentElement = elements.create("payment");
paymentElement.mount("#payment-element");

document.getElementById("pay-button").addEventListener("click", async () => {
  const { error } = await stripe.confirmPayment({
    elements,
    confirmParams: {
      return_url: "https://yourdomain.com/complete",
    },
  });

  if (error) {
    document.getElementById("errors").textContent = error.message;
  }
});

パターン 4：サブスクリプションの作成

def create_subscription(customer_id, price_id):
    """Create a subscription for a customer."""
    try:
        subscription = stripe.Subscription.create(
            customer=customer_id,
            items=[{'price': price_id}],
            payment_behavior='default_incomplete',
            payment_settings={'save_default_payment_method': 'on_subscription'},
            expand=['latest_invoice.payment_intent'],
        )

        return {
            'subscription_id': subscription.id,
            'client_secret': subscription.latest_invoice.payment_intent.client_secret
        }
    except stripe.error.StripeError as e:
        print(f"Subscription creation failed: {e}")
        raise

パターン 5：顧客ポータル

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

Webhook の処理

セキュアな Webhook エンドポイント

from flask import Flask, request
import stripe

app = Flask(__name__)

endpoint_secret = 'whsec_...'

@app.route('/webhook', methods=['POST'])
def webhook():
    payload = request.data
    sig_header = request.headers.get('Stripe-Signature')

    try:
        event = stripe.Webhook.construct_event(
            payload, sig_header, endpoint_secret
        )
    except ValueError:
        # Invalid payload
        return 'Invalid payload', 400
    except stripe.error.SignatureVerificationError:
        # Invalid signature
        return 'Invalid signature', 400

    # 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']}")

Webhook のベストプラクティス

import hashlib
import hmac

def verify_webhook_signature(payload, signature, secret):
    """Manually verify webhook signature."""
    expected_sig = hmac.new(
        secret.encode('utf-8'),
        payload,
        hashlib.sha256
    ).hexdigest()

    return hmac.compare_digest(signature, expected_sig)

def handle_webhook_idempotently(event_id, handler):
    """Ensure webhook is processed exactly once."""
    # Check if event already processed
    if is_event_processed(event_id):
        return

    # Process event
    try:
        handler()
        mark_event_processed(event_id)
    except Exception as e:
        log_error(e)
        # Stripe will retry failed webhooks
        raise

顧客管理

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, amount=None, reason=None):
    """Create a refund."""
    refund_params = {
        'payment_intent': payment_intent_id
    }

    if amount:
        refund_params['amount'] = amount  # Partial refund

    if reason:
        refund_params['reason'] = reason  # 'duplicate', 'fraudulent', 'requested_by_customer'

    refund = stripe.Refund.create(**refund_params)
    return refund

def handle_dispute(charge_id, evidence):
    """Update dispute with evidence."""
    stripe.Dispute.modify(
        charge_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
stripe.api_key = "sk_test_..."

# 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="[email protected]"
    )

    # Create payment intent
    intent = stripe.PaymentIntent.create(
        amount=1000,
        automatic_payment_methods={
            'enabled': True
        },
        currency='usd',
        customer=customer.id
    )

    # Confirm with test card
    confirmed = stripe.PaymentIntent.confirm(
        intent.id,
        payment_method='pm_card_visa'  # Test payment method
    )

    assert confirmed.status == 'succeeded'