Payment Gateway Integration Overview

Build secure payment processing systems with major payment providers (Stripe, PayPal, Square), handling transactions, subscriptions, webhooks, PCI compliance, and error scenarios across different backend frameworks.

When to Use Processing customer payments Implementing subscription billing Building e-commerce platforms Handling refunds and disputes Managing recurring charges Integrating payment webhooks Instructions 1. Stripe Integration with Python/Flask

config.py

import os

class StripeConfig: STRIPE_SECRET_KEY = os.getenv('STRIPE_SECRET_KEY') STRIPE_PUBLISHABLE_KEY = os.getenv('STRIPE_PUBLISHABLE_KEY') STRIPE_WEBHOOK_SECRET = os.getenv('STRIPE_WEBHOOK_SECRET')

stripe_service.py

import stripe from datetime import datetime, timedelta import logging

logger = logging.getLogger(name)

stripe.api_key = os.getenv('STRIPE_SECRET_KEY')

class StripePaymentService: @staticmethod def create_payment_intent(amount, currency='usd', description=None, metadata=None): """Create a payment intent""" try: intent = stripe.PaymentIntent.create( amount=int(amount * 100), # Convert to cents currency=currency, description=description, metadata=metadata or {} ) logger.info(f"Payment intent created: {intent.id}") return { 'success': True, 'client_secret': intent.client_secret, 'intent_id': intent.id } except stripe.error.StripeError as e: logger.error(f"Stripe error: {str(e)}") return {'success': False, 'error': str(e)}

@staticmethod
def confirm_payment(intent_id):
    """Confirm payment intent"""
    try:
        intent = stripe.PaymentIntent.retrieve(intent_id)

        if intent.status == 'succeeded':
            logger.info(f"Payment confirmed: {intent_id}")
            return {'success': True, 'intent_id': intent_id, 'status': intent.status}
        else:
            return {'success': False, 'status': intent.status}

    except stripe.error.StripeError as e:
        return {'success': False, 'error': str(e)}

@staticmethod
def create_customer(email, name=None, metadata=None):
    """Create Stripe customer"""
    try:
        customer = stripe.Customer.create(
            email=email,
            name=name,
            metadata=metadata or {}
        )
        logger.info(f"Customer created: {customer.id}")
        return {'success': True, 'customer_id': customer.id}
    except stripe.error.StripeError as e:
        return {'success': False, 'error': str(e)}

@staticmethod
def create_subscription(customer_id, price_id, metadata=None):
    """Create recurring subscription"""
    try:
        subscription = stripe.Subscription.create(
            customer=customer_id,
            items=[{'price': price_id}],
            metadata=metadata or {}
        )
        logger.info(f"Subscription created: {subscription.id}")
        return {
            'success': True,
            'subscription_id': subscription.id,
            'status': subscription.status
        }
    except stripe.error.StripeError as e:
        return {'success': False, 'error': str(e)}

@staticmethod
def cancel_subscription(subscription_id):
    """Cancel subscription"""
    try:
        subscription = stripe.Subscription.delete(subscription_id)
        logger.info(f"Subscription cancelled: {subscription_id}")
        return {'success': True, 'subscription_id': subscription_id}
    except stripe.error.StripeError as e:
        return {'success': False, 'error': str(e)}

@staticmethod
def refund_payment(payment_intent_id, amount=None):
    """Refund a payment"""
    try:
        refund = stripe.Refund.create(
            payment_intent=payment_intent_id,
            **({'amount': int(amount * 100)} if amount else {})
        )
        logger.info(f"Refund created: {refund.id}")
        return {'success': True, 'refund_id': refund.id}
    except stripe.error.StripeError as e:
        return {'success': False, 'error': str(e)}

routes.py

from flask import Blueprint, request, jsonify from stripe_service import StripePaymentService from functools import wraps import hmac import hashlib

payment_bp = Blueprint('payments', name, url_prefix='/api/payments')

def verify_stripe_webhook(f): @wraps(f) def decorated_function(args, *kwargs): signature = request.headers.get('Stripe-Signature') webhook_secret = os.getenv('STRIPE_WEBHOOK_SECRET')

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

    request.stripe_event = event
    return f(*args, **kwargs)

return decorated_function

@payment_bp.route('/create-intent', methods=['POST']) @token_required def create_payment_intent(): """Create payment intent""" data = request.json amount = data.get('amount') description = data.get('description')

if not amount or amount <= 0:
    return jsonify({'error': 'Invalid amount'}), 400

result = StripePaymentService.create_payment_intent(
    amount=amount,
    description=description,
    metadata={'user_id': current_user.id}
)

if result['success']:
    return jsonify(result), 200
else:
    return jsonify(result), 400

@payment_bp.route('/confirm-payment', methods=['POST']) @token_required def confirm_payment(): """Confirm payment""" data = request.json intent_id = data.get('intent_id')

result = StripePaymentService.confirm_payment(intent_id)

if result['success']:
    # Update user's payment status in database
    order = Order.query.filter_by(
        stripe_intent_id=intent_id,
        user_id=current_user.id
    ).first()

    if order:
        order.status = 'paid'
        db.session.commit()

    return jsonify(result), 200
else:
    return jsonify(result), 400

@payment_bp.route('/subscribe', methods=['POST']) @token_required def create_subscription(): """Create subscription""" data = request.json price_id = data.get('price_id')

if not price_id:
    return jsonify({'error': 'Price ID required'}), 400

# Get or create Stripe customer
user = current_user
if not user.stripe_customer_id:
    customer_result = StripePaymentService.create_customer(
        email=user.email,
        name=user.full_name
    )
    if not customer_result['success']:
        return jsonify(customer_result), 400
    user.stripe_customer_id = customer_result['customer_id']
    db.session.commit()

result = StripePaymentService.create_subscription(
    customer_id=user.stripe_customer_id,
    price_id=price_id,
    metadata={'user_id': user.id}
)

if result['success']:
    subscription = Subscription(
        user_id=user.id,
        stripe_subscription_id=result['subscription_id'],
        status=result['status']
    )
    db.session.add(subscription)
    db.session.commit()
    return jsonify(result), 201
else:
    return jsonify(result), 400

@payment_bp.route('/webhook', methods=['POST']) @verify_stripe_webhook def handle_webhook(): """Handle Stripe webhooks""" event = request.stripe_event

try:
    if event['type'] == 'payment_intent.succeeded':
        intent = event['data']['object']
        logger.info(f"Payment succeeded: {intent['id']}")
        # Update order status

    elif event['type'] == 'payment_intent.payment_failed':
        intent = event['data']['object']
        logger.error(f"Payment failed: {intent['id']}")
        # Handle failed payment

    elif event['type'] == 'customer.subscription.updated':
        subscription = event['data']['object']
        logger.info(f"Subscription updated: {subscription['id']}")

    elif event['type'] == 'customer.subscription.deleted':
        subscription = event['data']['object']
        logger.info(f"Subscription deleted: {subscription['id']}")
        # Update user's subscription status

    elif event['type'] == 'invoice.payment_succeeded':
        invoice = event['data']['object']
        logger.info(f"Invoice paid: {invoice['id']}")

    elif event['type'] == 'invoice.payment_failed':
        invoice = event['data']['object']
        logger.error(f"Invoice payment failed: {invoice['id']}")

    return jsonify({'received': True}), 200

except Exception as e:
    logger.error(f"Webhook error: {str(e)}")
    return jsonify({'error': str(e)}), 500

Node.js/Express Stripe Integration // stripe-service.js const stripe = require('stripe')(process.env.STRIPE_SECRET_KEY); const logger = require('./logger');

class StripeService { async createPaymentIntent(amount, currency = 'usd', metadata = {}) { try { const intent = await stripe.paymentIntents.create({ amount: Math.round(amount * 100), currency: currency, metadata: metadata });

        logger.info(`Payment intent created: ${intent.id}`);
        return { success: true, clientSecret: intent.client_secret, intentId: intent.id };
    } catch (error) {
        logger.error(`Stripe error: ${error.message}`);
        return { success: false, error: error.message };
    }
}

async createCustomer(email, name, metadata = {}) {
    try {
        const customer = await stripe.customers.create({
            email: email,
            name: name,
            metadata: metadata
        });

        logger.info(`Customer created: ${customer.id}`);
        return { success: true, customerId: customer.id };
    } catch (error) {
        return { success: false, error: error.message };
    }
}

async createSubscription(customerId, priceId, metadata = {}) {
    try {
        const subscription = await stripe.subscriptions.create({
            customer: customerId,
            items: [{ price: priceId }],
            metadata: metadata
        });

        logger.info(`Subscription created: ${subscription.id}`);
        return { success: true, subscriptionId: subscription.id, status: subscription.status };
    } catch (error) {
        return { success: false, error: error.message };
    }
}

async cancelSubscription(subscriptionId) {
    try {
        await stripe.subscriptions.del(subscriptionId);
        logger.info(`Subscription cancelled: ${subscriptionId}`);
        return { success: true };
    } catch (error) {
        return { success: false, error: error.message };
    }
}

async refundPayment(paymentIntentId, amount = null) {
    try {
        const refund = await stripe.refunds.create({
            payment_intent: paymentIntentId,
            ...(amount && { amount: Math.round(amount * 100) })
        });

        logger.info(`Refund created: ${refund.id}`);
        return { success: true, refundId: refund.id };
    } catch (error) {
        return { success: false, error: error.message };
    }
}

}

module.exports = new StripeService();

// routes.js const express = require('express'); const stripe = require('stripe')(process.env.STRIPE_SECRET_KEY); const stripeService = require('../services/stripe-service'); const { authenticate } = require('../middleware/auth');

const router = express.Router();

router.post('/create-intent', authenticate, async (req, res) => { const { amount, description } = req.body;

if (!amount || amount <= 0) {
    return res.status(400).json({ error: 'Invalid amount' });
}

const result = await stripeService.createPaymentIntent(amount, 'usd', {
    userId: req.user.id,
    description: description
});

if (result.success) {
    res.json(result);
} else {
    res.status(400).json(result);
}

});

router.post('/webhook', express.raw({ type: 'application/json' }), async (req, res) => { const signature = req.headers['stripe-signature'];

try {
    const event = stripe.webhooks.constructEvent(
        req.body,
        signature,
        process.env.STRIPE_WEBHOOK_SECRET
    );

    if (event.type === 'payment_intent.succeeded') {
        const intent = event.data.object;
        logger.info(`Payment succeeded: ${intent.id}`);
        // Update order status

    } else if (event.type === 'customer.subscription.updated') {
        const subscription = event.data.object;
        logger.info(`Subscription updated: ${subscription.id}`);

    } else if (event.type === 'invoice.payment_succeeded') {
        const invoice = event.data.object;
        logger.info(`Invoice paid: ${invoice.id}`);
    }

    res.json({ received: true });
} catch (error) {
    logger.error(`Webhook error: ${error.message}`);
    res.status(400).send(`Webhook Error: ${error.message}`);
}

});

module.exports = router;

PayPal Integration

paypal_service.py

import paypalrestsdk import os

paypalrestsdk.configure({ "mode": os.getenv("PAYPAL_MODE", "sandbox"), "client_id": os.getenv("PAYPAL_CLIENT_ID"), "client_secret": os.getenv("PAYPAL_CLIENT_SECRET") })

class PayPalService: @staticmethod def create_payment(amount, currency='USD', return_url=None, cancel_url=None): """Create PayPal payment""" payment = paypalrestsdk.Payment({ "intent": "sale", "payer": { "payment_method": "paypal" }, "redirect_urls": { "return_url": return_url or "https://example.com/return", "cancel_url": cancel_url or "https://example.com/cancel" }, "transactions": [{ "amount": { "total": str(amount), "currency": currency, "details": { "subtotal": str(amount) } }, "description": "Payment" }] })

    if payment.create():
        logger.info(f"PayPal payment created: {payment.id}")
        approval_url = None
        for link in payment.links:
            if link['rel'] == 'approval_url':
                approval_url = link['href']

        return {
            'success': True,
            'payment_id': payment.id,
            'approval_url': approval_url
        }
    else:
        logger.error(f"PayPal error: {payment.error}")
        return {'success': False, 'error': payment.error}

@staticmethod
def execute_payment(payment_id, payer_id):
    """Execute approved payment"""
    payment = paypalrestsdk.Payment.find(payment_id)

    if payment.execute({"payer_id": payer_id}):
        logger.info(f"Payment executed: {payment.id}")
        return {'success': True, 'transaction_id': payment.transactions[0].related_resources[0].sale.id}
    else:
        logger.error(f"Execution error: {payment.error}")
        return {'success': False, 'error': payment.error}

Subscription Management

subscription_service.py

class SubscriptionService: @staticmethod def create_subscription(user_id, plan_id, payment_method_id): """Create user subscription""" try: result = StripePaymentService.create_subscription( customer_id=user.stripe_customer_id, price_id=plan_id )

        if result['success']:
            subscription = Subscription(
                user_id=user_id,
                stripe_subscription_id=result['subscription_id'],
                plan_id=plan_id,
                status='active',
                started_at=datetime.utcnow(),
                renewal_date=datetime.utcnow() + timedelta(days=30)
            )
            db.session.add(subscription)
            db.session.commit()

            logger.info(f"Subscription created for user {user_id}")
            return {'success': True, 'subscription_id': subscription.id}

    except Exception as e:
        logger.error(f"Failed to create subscription: {str(e)}")
        return {'success': False, 'error': str(e)}

@staticmethod
def cancel_subscription(subscription_id):
    """Cancel subscription"""
    subscription = Subscription.query.get(subscription_id)
    if not subscription:
        return {'success': False, 'error': 'Subscription not found'}

    result = StripePaymentService.cancel_subscription(subscription.stripe_subscription_id)

    if result['success']:
        subscription.status = 'cancelled'
        subscription.cancelled_at = datetime.utcnow()
        db.session.commit()

        logger.info(f"Subscription cancelled: {subscription_id}")
        return {'success': True}

    return result

Best Practices ✅ DO Use official payment SDK libraries Verify webhook signatures Store minimal payment information Never store full credit card numbers Use HTTPS for all payment routes Implement proper error handling Test with sandbox environments Handle payment failures gracefully Implement PCI compliance Log all payment transactions Use idempotency keys Implement retry logic ❌ DON'T Handle raw card data Store sensitive payment information Log sensitive details Trust client-side validation only Ignore webhook events Hardcode API keys Use test keys in production Skip SSL/TLS verification Forget to validate amounts Store payment tokens without encryption Complete Example @app.post("/pay") async def process_payment(amount: float, current_user: dict = Depends(get_current_user)): result = StripePaymentService.create_payment_intent(amount) if result['success']: return result raise HTTPException(status_code=400, detail=result['error'])

payment-gateway-integration

安装

config.py

stripe_service.py

routes.py

paypal_service.py

subscription_service.py