Stripe Payment Integration: Complete Developer Guide

pip install stripe
export STRIPE_SECRET_KEY=sk_test_...
export STRIPE_PUBLISHABLE_KEY=pk_test_...

# payment_intent.py
# Create and confirm a PaymentIntent

import stripe
import os

stripe.api_key = os.environ['STRIPE_SECRET_KEY']

def create_payment(amount=2000, currency='usd'):
    """Create a PaymentIntent for a one-time payment.
    
    Args:
        amount: Amount in cents ($20.00 = 2000)
        currency: ISO currency code
    """
    intent = stripe.PaymentIntent.create(
        amount=amount,
        currency=currency,
        automatic_payment_methods={'enabled': True},
        description='Pro Plan Annual Subscription',
        metadata={
            'order_id': 'ORD-12345',
            'customer_email': 'alice@example.com'
        },
    )

    print(f"PaymentIntent created: {intent.id}")
    print(f"Amount: ${intent.amount / 100:.2f}")
    print(f"Status: {intent.status}")
    print(f"Client Secret: {intent.client_secret[:20]}...")

    return intent

# Test
intent = create_payment(2999, 'usd')

# checkout_session.py
# Create a Stripe Checkout Session for subscription

import stripe
import os

stripe.api_key = os.environ['STRIPE_SECRET_KEY']

def create_checkout_session():
    """Create a Stripe Checkout Session for subscription purchase."""
    session = stripe.checkout.Session.create(
        success_url='https://myapp.com/success?session_id={CHECKOUT_SESSION_ID}',
        cancel_url='https://myapp.com/pricing',
        mode='subscription',
        line_items=[{
            'price': 'price_1MqwertyABC123',  # Create this in Stripe Dashboard
            'quantity': 1,
        }],
        customer_email='alice@example.com',
        metadata={
            'plan': 'pro_annual',
            'source': 'landing_page',
        },
        subscription_data={
            'metadata': {'plan_tier': 'pro'},
            'trial_period_days': 14,
        },
        allow_promotion_codes=True,
        automatic_tax={'enabled': True},
    )

    print(f"Checkout Session: {session.id}")
    print(f"URL: {session.url}")
    print(f"Mode: {session.mode}")

    # Redirect user to session.url
    return session

session = create_checkout_session()

# webhooks.py
# Stripe webhook handler for payment events

from flask import Flask, request, jsonify
import stripe
import os
import json

app = Flask(__name__)
stripe.api_key = os.environ['STRIPE_SECRET_KEY']
endpoint_secret = os.environ['STRIPE_WEBHOOK_SECRET']

@app.route('/stripe/webhook', methods=['POST'])
def stripe_webhook():
    """Handle Stripe webhook events."""
    payload = request.get_data(as_text=True)
    sig_header = request.headers.get('Stripe-Signature')

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

    # Handle the event
    event_type = event['type']
    data = event['data']['object']

    print(f"Received event: {event_type}")

    handlers = {
        'checkout.session.completed': handle_checkout_completed,
        'payment_intent.succeeded': handle_payment_succeeded,
        'payment_intent.payment_failed': handle_payment_failed,
        'customer.subscription.updated': handle_subscription_updated,
        'customer.subscription.deleted': handle_subscription_deleted,
    }

    handler = handlers.get(event_type)
    if handler:
        handler(data)
    else:
        print(f"Unhandled event type: {event_type}")

    return jsonify({'status': 'ok'}), 200

def handle_checkout_completed(session):
    """Fulfill the order after successful checkout."""
    customer_email = session.get('customer_details', {}).get('email')
    subscription_id = session.get('subscription')
    metadata = session.get('metadata', {})

    print(f"Checkout completed for {customer_email}")
    print(f"Subscription: {subscription_id}")
    print(f"Plan: {metadata.get('plan')}")

    # Activate account, send welcome email, etc.
    # update_user_subscription(customer_email, subscription_id)

def handle_payment_succeeded(payment_intent):
    """Process successful payment."""
    amount = payment_intent['amount'] / 100
    currency = payment_intent['currency'].upper()
    print(f"Payment received: ${amount:.2f} {currency}")

def handle_payment_failed(payment_intent):
    """Notify customer of payment failure."""
    customer_email = payment_intent.get('receipt_email', 'unknown')
    print(f"Payment failed for {customer_email}")
    # Send notification, ask to update payment method

def handle_subscription_updated(subscription):
    """Handle subscription status changes."""
    status = subscription['status']
    customer = subscription['customer']
    print(f"Subscription {subscription['id']}: {status} for customer {customer}")

def handle_subscription_deleted(subscription):
    """Handle subscription cancellation."""
    print(f"Subscription {subscription['id']} cancelled")
    # Downgrade account, archive data, etc.

if __name__ == '__main__':
    # Test locally: stripe listen --forward-to localhost:5000/stripe/webhook
    app.run(port=5000, debug=True)

# refunds.py
# Process full and partial refunds

import stripe
import os

stripe.api_key = os.environ['STRIPE_SECRET_KEY']

def process_full_refund(payment_intent_id, reason='requested_by_customer'):
    """Process a full refund for a PaymentIntent."""
    try:
        refund = stripe.Refund.create(
            payment_intent=payment_intent_id,
            reason=reason,
            metadata={
                'refund_initiated_by': 'support_agent',
                'ticket_id': 'TKT-789',
            }
        )
        print(f"Refund processed: {refund.id}")
        print(f"Amount: ${refund.amount / 100:.2f}")
        print(f"Status: {refund.status}")
        return refund
    except stripe.error.InvalidRequestError as e:
        print(f"Refund failed: {e}")
        return None

def process_partial_refund(payment_intent_id, amount_cents):
    """Process a partial refund for a specific amount."""
    refund = stripe.Refund.create(
        payment_intent=payment_intent_id,
        amount=amount_cents,
        reason='requested_by_customer',
    )
    print(f"Partial refund of ${amount_cents / 100:.2f}: {refund.id}")
    return refund

# Test
process_full_refund('pi_3Mqwerty19QfG5XG0gTzFz1L')

# subscriptions.py
# Manage Stripe subscriptions: create, update, cancel

import stripe
import os

stripe.api_key = os.environ['STRIPE_SECRET_KEY']

def list_active_subscriptions():
    """List all active subscriptions."""
    subscriptions = stripe.Subscription.list(status='active', limit=10)
    for sub in subscriptions.auto_paging_iter():
        print(f"{sub.id}: {sub.customer} — "
              f"${sub.plan.amount / 100:.2f}/{sub.plan.interval}")
    return subscriptions

def update_subscription_plan(subscription_id, new_price_id):
    """Upgrade/downgrade a subscription to a new price."""
    subscription = stripe.Subscription.retrieve(subscription_id)
    updated = stripe.Subscription.modify(
        subscription_id,
        items=[{
            'id': subscription['items']['data'][0]['id'],
            'price': new_price_id,
        }],
        proration_behavior='always_invoice',  # Charge/credit prorated amount
    )
    print(f"Subscription {subscription_id} updated to price {new_price_id}")
    return updated

def cancel_subscription(subscription_id, at_period_end=True):
    """Cancel a subscription."""
    if at_period_end:
        sub = stripe.Subscription.modify(
            subscription_id,
            cancel_at_period_end=True,
        )
        print(f"Subscription {subscription_id} will cancel at period end")
    else:
        sub = stripe.Subscription.delete(subscription_id)
        print(f"Subscription {subscription_id} cancelled immediately")

    return sub

def reactivate_subscription(subscription_id):
    """Reactivate a subscription set to cancel."""
    sub = stripe.Subscription.modify(
        subscription_id,
        cancel_at_period_end=False,
    )
    print(f"Subscription {subscription_id} reactivated")
    return sub

# Usage
# cancel_subscription('sub_1MqwertyABC123', at_period_end=True)

# idempotency_test.py
# Use idempotency keys to prevent duplicate charges

import stripe
import os
import uuid

stripe.api_key = os.environ['STRIPE_SECRET_KEY']

def charge_with_idempotency(amount, currency='usd'):
    """Create a payment with idempotency to prevent duplicates."""
    idempotency_key = str(uuid.uuid4())

    try:
        intent = stripe.PaymentIntent.create(
            amount=amount,
            currency=currency,
            confirm=True,
            payment_method='pm_card_visa',
            idempotency_key=idempotency_key,
        )
        print(f"Payment created: {intent.id} (key: {idempotency_key})")

        # If we retry with the same key, Stripe returns the same result
        duplicate = stripe.PaymentIntent.create(
            amount=amount,
            currency=currency,
            confirm=True,
            payment_method='pm_card_visa',
            idempotency_key=idempotency_key,  # Same key!
        )
        print(f"Duplicate result: {duplicate.id} (same as original)")

        return intent
    except stripe.error.StripeError as e:
        print(f"Stripe error: {e}")
        return None

charge_with_idempotency(1000)

# dashboard.py
# Monitor Stripe account metrics

import stripe
import os
from datetime import datetime, timedelta

stripe.api_key = os.environ['STRIPE_SECRET_KEY']

def stripe_dashboard():
    """Print current Stripe account metrics."""
    print("═" * 50)
    print("Stripe Account Dashboard")
    print("═" * 50)

    # Balance
    balance = stripe.Balance.retrieve()
    available = sum(b['amount'] for b in balance['available'])
    pending = sum(b['amount'] for b in balance['pending'])
    print(f"\nBalance:")
    print(f"  Available: ${available / 100:.2f}")
    print(f"  Pending:   ${pending / 100:.2f}")

    # Recent charges
    charges = stripe.Charge.list(limit=5)
    print(f"\nRecent Charges (last 5):")
    for charge in charges:
        created = datetime.fromtimestamp(charge['created'])
        print(f"  {charge['id'][:20]:<22} "
              f"${charge['amount'] / 100:>7.2f} "
              f"{charge['currency'].upper():<4} "
              f"{'✅' if charge['paid'] else '❌'} "
              f"{created.strftime('%m/%d %H:%M')}")

    # Active subscriptions
    subs = stripe.Subscription.list(status='active', limit=1)
    total_subs = 0
    for _ in subs.auto_paging_iter():
        total_subs += 1
        if total_subs >= 1000:
            break
    print(f"\nActive subscriptions: {total_subs}+")
    print("═" * 50)

if __name__ == '__main__':
    stripe_dashboard()