Payment Integration: Stripe & PayPal Implementation

Problem

Modern e-commerce applications require secure, reliable payment processing. Developers face challenges integrating multiple payment gateways while maintaining PCI compliance, handling various payment methods, managing transaction states, and ensuring data security throughout the payment flow.

Solution

Implement a unified payment abstraction layer supporting both Stripe and PayPal with:

• Secure token-based transactions
• Comprehensive error handling
• Transaction state management
• Webhook verification
• PCI compliance through tokenization
• Idempotency for retry safety

---

Code Implementation

1. Environment Configuration

# .env
STRIPE_SECRET_KEY=sk_live_xxxxx
STRIPE_PUBLISHABLE_KEY=pk_live_xxxxx
PAYPAL_CLIENT_ID=xxxxx
PAYPAL_SECRET=xxxxx
PAYPAL_MODE=sandbox
WEBHOOK_SECRET_STRIPE=whsec_xxxxx
WEBHOOK_SECRET_PAYPAL=xxxxx

2. Payment Gateway Abstraction

# payment_gateway.py
from abc import ABC, abstractmethod
from typing import Dict, Optional
from dataclasses import dataclass
from enum import Enum
import logging

logger = logging.getLogger(__name__)

class PaymentStatus(Enum):
    PENDING = "pending"
    PROCESSING = "processing"
    COMPLETED = "completed"
    FAILED = "failed"
    REFUNDED = "refunded"
    CANCELLED = "cancelled"

@dataclass
class PaymentRequest:
    amount: float
    currency: str
    customer_id: str
    description: str
    metadata: Dict = None
    idempotency_key: str = None

@dataclass
class PaymentResponse:
    transaction_id: str
    status: PaymentStatus
    amount: float
    currency: str
    timestamp: str
    raw_response: Dict = None

class PaymentGateway(ABC):
    """Abstract base class for payment gateways"""

    @abstractmethod
    def charge(self, request: PaymentRequest, token: str) -> PaymentResponse:
        pass

    @abstractmethod
    def refund(self, transaction_id: str, amount: Optional[float] = None) -> PaymentResponse:
        pass

    @abstractmethod
    def verify_webhook(self, payload: bytes, signature: str) -> bool:
        pass

    @abstractmethod
    def parse_webhook(self, payload: Dict) -> Dict:
        pass

3. Stripe Implementation

# stripe_gateway.py
import stripe
import hmac
import hashlib
from datetime import datetime
from payment_gateway import PaymentGateway, PaymentRequest, PaymentResponse, PaymentStatus
import os

class StripeGateway(PaymentGateway):
    def __init__(self):
        stripe.api_key = os.getenv('STRIPE_SECRET_KEY')
        self.webhook_secret = os.getenv('WEBHOOK_SECRET_STRIPE')

    def charge(self, request: PaymentRequest, token: str) -> PaymentResponse:
        """Process payment through Stripe"""
        try:
            # Create idempotency key for retry safety
            idempotency_key = request.idempotency_key or f"{request.customer_id}_{datetime.utcnow().timestamp()}"

            charge = stripe.Charge.create(
                amount=int(request.amount * 100),  # Convert to cents
                currency=request.currency,
                source=token,
                description=request.description,
                metadata={
                    'customer_id': request.customer_id,
                    **(request.metadata or {})
                },
                idempotency_key=idempotency_key
            )

            logger.info(f"Stripe charge created: {charge.id}")

            return PaymentResponse(
                transaction_id=charge.id,
                status=PaymentStatus.COMPLETED if charge.paid else PaymentStatus.FAILED,
                amount=charge.amount / 100,
                currency=charge.currency.upper(),
                timestamp=datetime.fromtimestamp(charge.created).isoformat(),
                raw_response=charge
            )

        except stripe.error.CardError as e:
            logger.error(f"Card error: {e.user_message}")
            return PaymentResponse(
                transaction_id="",
                status=PaymentStatus.FAILED,
                amount=request.amount,
                currency=request.currency,
                timestamp=datetime.utcnow().isoformat(),
                raw_response={'error': str(e)}
            )

        except stripe.error.RateLimitError:
            logger.error("Stripe rate limit exceeded")
            raise

        except stripe.error.InvalidRequestError as e:
            logger.error(f"Invalid request: {e}")
            raise

    def refund(self, transaction_id: str, amount: Optional[float] = None) -> PaymentResponse:
        """Refund a Stripe charge"""
        try:
            refund_params = {'charge': transaction_id}
            if amount:
                refund_params['amount'] = int(amount * 100)

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

            logger.info(f"Refund created: {refund.id}")

            return PaymentResponse(
                transaction_id=refund.id,
                status=PaymentStatus.REFUNDED,
                amount=refund.amount / 100,
                currency=refund.currency.upper(),
                timestamp=datetime.fromtimestamp(refund.created).isoformat(),
                raw_response=refund
            )

        except stripe.error.InvalidRequestError as e:
            logger.error(f"Refund failed: {e}")
            raise

    def verify_webhook(self, payload: bytes, signature: str) -> bool:
        """Verify Stripe webhook signature"""
        try:
            computed_signature = hmac.new(
                self.webhook_secret.encode(),
                payload,
                hashlib.sha256
            ).hexdigest()

            return hmac.compare_digest(computed_signature, signature)
        except Exception as e:
            logger.error(f"Webhook verification failed: {e}")
            return False

    def parse_webhook(self, payload: Dict) -> Dict:
        """Parse Stripe webhook event"""
        event_type = payload.get('type')
        data = payload.get('data', {}).get('object', {})

        webhook_map = {
            'charge.succeeded': {
                'event': 'payment_completed',
                'transaction_id': data.get('id'),
                'status': PaymentStatus.COMPLETED,
                'amount': data.get('amount') / 100,
                'customer_id': data.get('metadata', {}).get('customer_id')
            },
            'charge.failed': {
                'event': 'payment_failed',
                'transaction_id': data.get('id'),
                'status': PaymentStatus.FAILED,
                'amount': data.get('amount') / 100,
                'customer_id': data.get('metadata', {}).get('customer_id')
            },
            'charge.refunded': {
                'event': 'payment_refunded',
                'transaction_id': data.get('id'),
                'status': PaymentStatus.REFUNDED,
                'amount': data.get('amount_refunded') / 100,
                'customer_id': data.get('metadata', {}).get('customer_id')
            }
        }

        return webhook_map.get(event_type, {})

4. PayPal Implementation

# paypal_gateway.py
import requests
import hmac
import hashlib
import base64
from datetime import datetime
from payment_gateway import PaymentGateway, PaymentRequest, PaymentResponse, PaymentStatus
import os
import logging

logger = logging.getLogger(__name__)

class PayPalGateway(PaymentGateway):
    def __init__(self):
        self.client_id = os.getenv('PAYPAL_CLIENT_ID')
        self.secret = os.getenv('PAYPAL_SECRET')
        self.mode = os.getenv('PAYPAL_MODE', 'sandbox')
        self.base_url = f"https://api.{self.mode}.paypal.com"
        self.webhook_secret = os.getenv('WEBHOOK_SECRET_PAYPAL')
        self.access_token = None

    def _get_access_token(self) -> str:
        """Obtain PayPal OAuth access token"""
        if self.access_token:
            return self.access_token

        auth = base64.b64encode(
            f"{self.client_id}:{self.secret}".encode()
        ).decode()

        headers = {
            'Authorization': f'Basic {auth}',
            'Content-Type': 'application/x-www-form-urlencoded'
        }

        response = requests.post(
            f"{self.base_url}/v1/oauth2/token",
            headers=headers,
            data={'grant_type': 'client_credentials'}
        )

        if response.status_code == 200:
            self.access_token = response.json()['access_token']
            return self.access_token

        raise Exception(f"Failed to get PayPal token: {response.text}")

    def charge(self, request: PaymentRequest, token: str) -> PaymentResponse:
        """Process payment through PayPal"""
        try:
            access_token = self._get_access_token()

            headers = {
                'Authorization': f'Bearer {access_token}',
                'Content-Type': 'application/json'
            }

            payload = {
                'intent': 'sale',
                'payer': {
                    'payment_method': 'credit_card',
                    'funding_instruments': [{
                        'credit_card_token': {
                            'token': token
                        }
                    }]
                },
                'transactions': [{
                    'amount': {
                        'total': str(request.amount),
                        'currency': request.currency,
                        'details': {
                            'subtotal': str(request.amount)
                        }
                    },
                    'description': request.description,
                    'custom': request.customer_id,
                    'invoice_number': request.idempotency_key or f"INV-{datetime.utcnow().timestamp()}"
                }]
            }

            response = requests.post(
                f"{self.base_url}/v1/payments/payment",
                json=payload,
                headers=headers
            )

            if response.status_code in [200, 201]:
                data = response.json()
                transaction_id = data['id']
                state = data.get('state', 'failed')

                logger.info(f"PayPal payment created: {transaction_id}")

                return PaymentResponse(
                    transaction_id=transaction_id,
                    status=PaymentStatus.COMPLETED if state == 'approved' else PaymentStatus.FAILED,
                    amount=request.amount,
                    currency=request.currency,
                    timestamp=datetime.utcnow().isoformat(),
                    raw_response=data
                )
            else:
                logger.error(f"PayPal error: {response.text}")
                return PaymentResponse(
                    transaction_id="",
                    status=PaymentStatus.FAILED,
                    amount=request.amount,
                    currency=request.currency,
                    timestamp=datetime.utcnow().isoformat(),
                    raw_response={'error': response.json()}
                )

        except Exception as e:
            logger.error(f"PayPal charge failed: {e}")
            raise

    def refund(self, transaction_id: str, amount: Optional[float] = None) -> PaymentResponse:
        """Refund a PayPal payment"""
        try:
            access_token = self._get_access_token()

            headers = {
                'Authorization': f'Bearer {access_token}',
                'Content-Type': 'application/json'
            }

            payload = {}
            if amount:
                payload['amount'] = {
                    'currency': 'USD',
                    'total': str(amount)
                }

            response = requests.post(
                f"{self.base_url}/v1/payments/payment/{transaction_id}/execute",
                json=payload,
                headers=headers
            )

            if response.status_code in [200, 201]:
                data = response.json()
                logger.info(f"PayPal refund processed: {transaction_id}")

                return PaymentResponse(
                    transaction_id=transaction_id,
                    status=PaymentStatus.REFUNDED,
                    amount=amount or 0,
                    currency='USD',
                    timestamp=datetime.utcnow().isoformat(),
                    raw_response=data
                )
            else:
                raise Exception(f"Refund failed: {response.text}")

        except Exception as e:
            logger.error(f"PayPal refund failed: {e}")
            raise

    def verify_webhook(self, payload: bytes, signature: str) -> bool:
        """Verify PayPal webhook signature"""
        try:
            computed_signature = hmac.new(
                self.webhook_secret.encode(),
                payload,
                hashlib.sha256
            ).digest()

            provided_signature = base64.b64decode(signature)

            return hmac.compare_digest(computed_signature, provided_signature)
        except Exception as e:
            logger.error(f"Webhook verification failed: {e}")
            return False

    def parse_webhook(self, payload: Dict) -> Dict:
        """Parse PayPal webhook event"""
        event_type = payload.get('event_type')
        resource = payload.get('resource', {})

        webhook_map = {
            'PAYMENT.SALE.COMPLETED': {
                'event': 'payment_completed',
                'transaction_id': resource.get('id'),
                'status': PaymentStatus.COMPLETED,
                'amount': float(resource.get('amount', {}).get('total', 0)),
                'customer_id': resource.get('custom')
            },
            'PAYMENT.SALE.DENIED': {
                'event': 'payment_failed',
                'transaction_id': resource.get('id'),
                'status': PaymentStatus.FAILED,
                'amount': float(resource.get('amount', {}).get('total', 0)),
                'customer_id': resource.get('custom')
            },
            'PAYMENT.SALE.REFUNDED': {
                'event': 'payment_refunded',
                'transaction_id': resource.get('id'),
                'status': PaymentStatus.REFUNDED,
                'amount': float(resource.get('amount', {}).get('total', 0)),
                'customer_id': resource.get('custom')
            }
        }

        return webhook_map.get(event_type, {})

5. Payment Service Layer

# payment_service.py
from typing import Optional, Dict
from payment_gateway import PaymentGateway, PaymentRequest, PaymentResponse, PaymentStatus
from stripe_gateway import StripeGateway
from paypal_gateway import PayPalGateway
import logging
from datetime import datetime

logger = logging.getLogger(__name__)

class PaymentService:
    def __init__(self):
        self.stripe = StripeGateway()
        self.paypal = PayPalGateway()
        self.transaction_log = {}  # In production, use database

    def process_payment(
        self,
        provider: str,
        amount: float,
        currency: str,
        customer_id: str,
        token: str,
        description: str,
        metadata: Optional[Dict] = None,
        idempotency_key: Optional[str] = None
    ) -> PaymentResponse:
        """Process payment with specified provider"""

        gateway = self._get_gateway(provider)

        request = PaymentRequest(
            amount=amount,
            currency=currency,
            customer_id=customer_id,
            description=description,
            metadata=metadata,
            idempotency_key=idempotency_key
        )

        try:
            response = gateway.charge(request, token)

            # Log transaction
            self._log_transaction(provider, response, request)

            return response

        except Exception as e:
            logger.error(f"Payment processing failed: {e}")
            raise

    def refund_payment(
        self,
        provider: str,
        transaction_id: str,
        amount: Optional[float] = None
    ) -> PaymentResponse:
        """Refund a payment"""

        gateway = self._get_gateway(provider)

        try:
            response = gateway.refund(transaction_id, amount)
            logger.info(f"Refund processed: {transaction_id}")
            return response

        except Exception as e:
            logger.error(f"Refund failed: {e}")
            raise

    def handle_webhook(
        self,
        provider: str,
        payload: bytes,
        signature: str
    ) -> Dict:
        """Handle payment provider webhook"""

        gateway = self._get_gateway(provider)

        # Verify signature
        if not gateway.verify_webhook(payload, signature):
            logger.warning(f"Invalid webhook signature from {provider}")
            raise ValueError("Invalid webhook signature")

        # Parse payload
        import json
        payload_dict = json.loads(payload)

        webhook_data = gateway.parse_webhook(payload_dict)

        logger.info(f"Webhook processed: {webhook_data.get('event')}")

        return webhook_data

    def _get_gateway(self, provider: str) -> PaymentGateway:
        """Get payment gateway instance"""
        if provider.lower() == 'stripe':
            return self.stripe
        elif provider.lower() == 'paypal':
            return self.paypal
        else:
            raise ValueError(f"Unknown payment provider: {provider}")

    def _log_transaction(
        self,
        provider: str,
        response: PaymentResponse,
        request: PaymentRequest
    ):
        """Log transaction for audit trail"""
        log_entry = {
            'provider': provider,
            'transaction_id': response.transaction_id,
            'status': response.status.value,
            'amount': response.amount,
            'currency': response.currency,
            'customer_id': request.customer_id,
            'timestamp': datetime.utcnow().isoformat()
        }

        self.transaction_log[response.transaction_id] = log_entry
        logger.info(f"Transaction logged: {log_entry}")

6. Flask API Endpoints

```python

app.py

from flask import Flask, request, jsonify from payment_service import PaymentService from functools import wraps import os

app = Flask(name) payment_service = PaymentService()

def require_api_key(f): """Decorator to verify API key""" @wraps(f) def decorated_function(args, **kwargs): api_key = request.headers.get('X-API-Key') if not api_key or api_key != os.getenv('API_KEY'): return jsonify({'error': 'Unauthorized'}), 401 return f(args, **kwargs) return decorated_function

@app.route('/api/payments/charge', methods=['POST']) @require_api_key def charge_payment(): """Process payment charge""" try: data