Complete Stripe-integrated payment system for handling subscriptions, one-time payments, and billing management

Overview

This payment system provides a complete solution for modern subscription-based applications:

Subscription Management: Recurring payments with flexible trial periods
One-time Payments: Credit packages and single purchases
Webhook Integration: Real-time payment event synchronization
Customer Portal: Self-service billing management
Multi-currency Support: Global payment processing
Credit System: Dual-credit approach with monthly and one-time allocations

This system is production-ready and handles millions of transactions with proper error handling, security, and monitoring.

Quick Setup

Follow these steps to get your payment system running quickly:

Create a free Stripe account
Complete account verification for production use
Navigate to Developers → API Keys in your dashboard
Copy your test API keys for development

Keep your API keys secure and never commit them to version control. Always use environment variables.

Create a .env file in your project root:

# Stripe Configuration
STRIPE_SECRET_KEY=sk_test_your_secret_key_here
STRIPE_PUBLIC_KEY=pk_test_your_publishable_key_here
WEBHOOK_SECRET=whsec_your_webhook_secret_here

# Application Configuration
FRONTEND_URL=http://localhost:3000
DATABASE_URL=postgresql://user:password@localhost:5432/your_db

# Optional: Enable debug logging
LOG_LEVEL=DEBUG

Security Critical: Never use test keys in production. Replace with live keys before deploying.

Install the required Python packages:

pip install -r requirements.txt

Required packages include:

stripe - Stripe Python SDK
fastapi - Web framework
sqlalchemy - Database ORM
alembic - Database migrations
pydantic - Data validation

Run database migrations to create the payment tables:

# Create migration (if not exists)
alembic revision --autogenerate -m "Add payment tables"

# Apply migrations
alembic upgrade head

This creates:

plans table for subscription products
packages table for one-time purchase products
Payment-related columns in the users table

Products must be created in Stripe before they can be used in your application.

Create Subscription Products

Go to Products → Add product in Stripe dashboard
Enter product details:
- Name: "Pro Plan"
- Pricing: Select "Recurring"
- Billing interval: Monthly/Yearly
- Trial period: 14 days (optional)
Save and copy the Product ID and Price ID

Create One-time Products

Go to Products → Add product
Enter product details:
- Name: "100 Credits"
- Pricing: Select "One-time"
- Price: Fixed amount
- Metadata: Add credits: 100 for tracking
Save and copy the Product ID and Price ID

Add Products to Database

After creating products in Stripe, add them to your database:

-- Subscription plan example
INSERT INTO plans (
    product_name,
    product_id,
    price_id,
    price,
    full_plan_name,
    full_price_name,
    monthly_uses
) VALUES (
    'Pro Plan',
    'prod_abc123',
    'price_def456',
    '29.99',
    'Professional Plan',
    'Pro Monthly',
    100
);

-- One-time package example
INSERT INTO packages (
    product_name,
    product_id,
    price_id,
    price,
    uses_amount
) VALUES (
    '100 Credits',
    'prod_xyz987',
    'price_ghi789',
    '19.99',
    100
);

Replace the Product IDs and Price IDs with actual values from your Stripe dashboard.

Database Models

Plan Model

The Plan model stores subscription plan information linked to Stripe products:

class Plan(Base):
    __tablename__ = "plans"

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    full_plan_name = Column(String, nullable=True, default="")
    full_price_name = Column(String, nullable=True, default="")
    product_name = Column(String, unique=True, nullable=False)
    product_id = Column(String, unique=True, nullable=False)  # Stripe product ID
    price_id = Column(String, unique=True, nullable=False)    # Stripe price ID
    price = Column(String, nullable=False)
    monthly_uses = Column(Integer, default=0)  # Monthly credit allocation

Package Model

The Package model stores one-time payment products:

class Package(Base):
    __tablename__ = "packages"

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    product_name = Column(String, unique=True, nullable=False)
    product_id = Column(String, unique=True, nullable=False)
    price_id = Column(String, unique=True, nullable=False)
    price = Column(String, nullable=False)
    uses_amount = Column(Integer, nullable=False)  # Credits to add

User Model Extensions

The payment system extends the User model with Stripe-related fields:

class User(Base):
    # ... existing fields

    # Stripe customer integration
    customer_id = Column(String(255), nullable=True)
    plan_id = Column(Integer, nullable=True)
    subscription_id = Column(String(255), nullable=True)
    subscription_status = Column(String(64), nullable=True)
    subscription_last_renew = Column(String, nullable=True)
    subscription_next_renew = Column(String, nullable=True)

    # Dual credit system
    remaining_post_creations = Column(Integer, default=0)  # Monthly credits
    remaining_post_creations_onetime_paid = Column(Integer, default=0)  # Purchased credits

The dual-credit system allows users to have both subscription-based monthly credits and purchased one-time credits, providing maximum flexibility.

API Reference

Product Management

Get Products

Retrieve available products filtered by type:

GET /payments/products/?product_type=subscription
GET /payments/products/?product_type=one_time

Parameters:

product_type: Either subscription or one_time

Response:

{
  "products": [
    {
      "id": "prod_abc123",
      "name": "Pro Plan",
      "description": "Professional subscription plan",
      "price": {
        "price_id": "price_def456",
        "unit_amount": "29.99",
        "currency": "usd",
        "recurring": {
          "interval": "month",
          "interval_count": 1
        }
      }
    }
  ]
}

Checkout Sessions

Create Subscription Checkout

POST /payments/create-checkout-session
Content-Type: application/json
Authorization: Bearer <access_token>

{
  "price_id": "price_def456",
  "product_id": "prod_abc123"
}

Features:

Automatic 14-day trial period
Customer creation if not exists
Subscription conflict detection

Create One-time Checkout

POST /payments/create-checkout-session-onetime
Content-Type: application/json
Authorization: Bearer <access_token>

{
  "price_id": "price_ghi789",
  "product_id": "prod_xyz987"
}

Features:

Immediate payment processing
Credit allocation on completion
Promotion code support

Subscription Management

Get User Subscription

GET /payments/get-user-subscription
Authorization: Bearer <access_token>

Returns detailed subscription information including plan details, billing cycles, and status.

Cancel Subscription

DELETE /payments/cancel-subscription
Authorization: Bearer <access_token>

Cancels subscription at the end of the current billing period (no immediate termination).

Billing Portal

GET /payments/create-billing-portal-session
Authorization: Bearer <access_token>

Creates a secure session for customers to manage their billing information, payment methods, and download invoices.

The billing portal is hosted by Stripe and provides a secure, PCI-compliant interface for customer self-service.

Webhook Integration

Webhooks are critical for keeping your application synchronized with Stripe events in real-time.

In Stripe Dashboard, go to Developers → Webhooks
Click Add endpoint
Set URL: https://yourdomain.com/payments/webhook
Select events to listen for:
- checkout.session.completed
- invoice.paid
- invoice.payment_failed
- customer.subscription.updated
- customer.subscription.deleted

Copy the webhook signing secret from Stripe and add it to your environment:

WEBHOOK_SECRET=whsec_your_webhook_secret_here

Security Critical: Always verify webhook signatures to prevent malicious requests from affecting your system.

Use Stripe CLI to test webhook events locally:

# Install and authenticate Stripe CLI
stripe login

# Forward webhooks to local server
stripe listen --forward-to localhost:8000/payments/webhook

# Test specific events
stripe trigger checkout.session.completed
stripe trigger invoice.paid
stripe trigger invoice.payment_failed

Webhook Event Handlers

The system handles these critical webhook events:

Checkout Session Completed

Processes new subscriptions and one-time purchases:

async def handle_checkout_completed(session: Dict[str, Any], db: Session):
    if session["mode"] == "subscription":
        # Create/update subscription
        # Assign monthly credits
        # Update user plan
    elif session["mode"] == "payment":
        # Process one-time purchase
        # Add credits to user account

Invoice Paid

Handles subscription renewals and credit resets:

async def handle_invoice_paid(session: Dict[str, Any], db: Session):
    # Reset monthly credits
    # Update renewal dates
    # Log successful payment

Payment Failed

Manages failed payments and subscription status:

async def handle_invoice_payment_failed(session: Dict[str, Any], db: Session):
    # Update subscription status
    # Trigger retry logic
    # Send notification to customer

Failed payments don't immediately cancel subscriptions. Stripe has built-in retry logic, and subscriptions are typically suspended after multiple failed attempts.

Frontend Integration

Product Display

// Fetch and display subscription plans
const fetchPlans = async () => {
  try {
    const response = await fetch(
      "/payments/products/?product_type=subscription",
      {
        headers: { Authorization: `Bearer ${getAuthToken()}` },
      }
    );

    if (!response.ok) throw new Error("Failed to fetch plans");

    const { products } = await response.json();
    return products;
  } catch (error) {
    console.error("Error fetching plans:", error);
    return [];
  }
};

// Display plans in UI
const displayPlans = async () => {
  const plans = await fetchPlans();

  plans.forEach((plan) => {
    const planElement = document.createElement("div");
    planElement.innerHTML = `
      <h3>${plan.name}</h3>
      <p>${plan.description}</p>
      <p>$${plan.price.unit_amount}/${plan.price.recurring.interval}</p>
      <button onclick="purchasePlan('${plan.price.price_id}', '${plan.id}')">
        Subscribe
      </button>
    `;
    document.getElementById("plans-container").appendChild(planElement);
  });
};

Checkout Flow

// Handle subscription purchase
const purchasePlan = async (priceId, productId) => {
  try {
    const response = await fetch("/payments/create-checkout-session", {
      method: "POST",
      headers: {
        Authorization: `Bearer ${getAuthToken()}`,
        "Content-Type": "application/json",
      },
      body: JSON.stringify({
        price_id: priceId,
        product_id: productId,
      }),
    });

    if (!response.ok) {
      const error = await response.json();
      throw new Error(error.detail || "Failed to create checkout session");
    }

    const { checkout_url } = await response.json();
    window.location.href = checkout_url;
  } catch (error) {
    console.error("Purchase error:", error);
    alert("Failed to start checkout process. Please try again.");
  }
};

// Handle one-time purchase
const purchaseCredits = async (priceId, productId) => {
  try {
    const response = await fetch("/payments/create-checkout-session-onetime", {
      method: "POST",
      headers: {
        Authorization: `Bearer ${getAuthToken()}`,
        "Content-Type": "application/json",
      },
      body: JSON.stringify({
        price_id: priceId,
        product_id: productId,
      }),
    });

    const { checkout_url } = await response.json();
    window.location.href = checkout_url;
  } catch (error) {
    console.error("Purchase error:", error);
    alert("Failed to purchase credits. Please try again.");
  }
};

Success and Error Handling

// Success page handler
const handlePaymentSuccess = () => {
  const urlParams = new URLSearchParams(window.location.search);
  const sessionId = urlParams.get("session_id");

  if (sessionId) {
    // Show success message
    showSuccessMessage("Payment successful! Your account has been updated.");

    // Redirect to dashboard after delay
    setTimeout(() => {
      window.location.href = "/dashboard";
    }, 3000);
  }
};

// Error handling
const handlePaymentError = (error) => {
  console.error("Payment error:", error);

  // Show user-friendly error message
  const errorMessages = {
    card_declined:
      "Your card was declined. Please try a different payment method.",
    insufficient_funds:
      "Insufficient funds. Please check your account balance.",
    expired_card: "Your card has expired. Please use a different card.",
    processing_error:
      "There was an error processing your payment. Please try again.",
  };

  const message =
    errorMessages[error.code] ||
    "An unexpected error occurred. Please try again.";
  showErrorMessage(message);
};

Credit Management System

The system uses a sophisticated dual-credit approach:

Credit Types

Monthly Credits (remaining_post_creations)
- Reset every billing cycle
- Provided by active subscriptions
- First priority for consumption
One-time Credits (remaining_post_creations_onetime_paid)
- Purchased separately
- Persist until used
- Used after monthly credits are exhausted

Credit Consumption Logic

def consume_credit(user: User, db: Session) -> bool:
    """
    Consume one credit from user's account.
    Returns True if credit was consumed, False if no credits available.
    """
    if user.remaining_post_creations > 0:
        user.remaining_post_creations -= 1
        db.commit()
        return True
    elif user.remaining_post_creations_onetime_paid > 0:
        user.remaining_post_creations_onetime_paid -= 1
        db.commit()
        return True
    else:
        return False

def get_total_credits(user: User) -> int:
    """Get user's total available credits"""
    return user.remaining_post_creations + user.remaining_post_creations_onetime_paid

def get_credit_breakdown(user: User) -> Dict[str, int]:
    """Get detailed credit breakdown"""
    return {
        "monthly_credits": user.remaining_post_creations,
        "purchased_credits": user.remaining_post_creations_onetime_paid,
        "total_credits": get_total_credits(user)
    }

Usage in Your Application

# Before allowing a user action that consumes credits
@router.post("/create-post")
async def create_post(user: user_dependency, db: db_dependency, post_data: CreatePostRequest):
    # Check if user has credits
    if not consume_credit(user, db):
        raise HTTPException(
            status_code=402,
            detail="No credits available. Please purchase credits or upgrade your plan."
        )

    # Proceed with post creation
    # ... your post creation logic here

    return {"message": "Post created successfully", "credits_remaining": get_total_credits(user)}

Testing

Test Environment Setup

Always use Stripe's test mode during development. No real money is processed, and you can simulate various scenarios.

Test Credit Cards

Use these test cards for different scenarios:

Card Number	Brand	Scenario
`4242424242424242`	Visa	Successful payment
`4000000000000002`	Visa	Declined payment
`4000000000000341`	Visa	Requires authentication
`4000002760003184`	Visa	Requires authentication (failure)
`5555555555554444`	Mastercard	Successful payment
`4000056655665556`	Visa Debit	Successful payment

Test Webhook Events

# Test checkout completion
stripe trigger checkout.session.completed

# Test subscription events
stripe trigger customer.subscription.created
stripe trigger invoice.paid
stripe trigger invoice.payment_failed

# Test one-time payment
stripe trigger payment_intent.succeeded

Integration Testing

# Example test for subscription creation
def test_create_subscription_checkout():
    # Setup test user
    user = create_test_user()

    # Create checkout session
    response = client.post(
        "/payments/create-checkout-session",
        json={"price_id": "price_test_123", "product_id": "prod_test_123"},
        headers={"Authorization": f"Bearer {get_test_token(user)}"}
    )

    assert response.status_code == 200
    assert "checkout_url" in response.json()

    # Verify customer was created
    updated_user = get_user_by_id(user.id)
    assert updated_user.customer_id is not None

Security Considerations

Webhook Security

Critical: Always verify webhook signatures to prevent malicious requests from affecting your system.

def verify_webhook_signature(payload: bytes, sig_header: str, endpoint_secret: str):
    """Verify Stripe webhook signature"""
    try:
        event = stripe.Webhook.construct_event(payload, sig_header, endpoint_secret)
        return event
    except ValueError:
        raise HTTPException(status_code=400, detail="Invalid payload")
    except stripe.error.SignatureVerificationError:
        raise HTTPException(status_code=400, detail="Invalid signature")

Input Validation

Use Pydantic models for robust request validation:

from pydantic import BaseModel, validator

class CreateCheckoutRequest(BaseModel):
    price_id: str
    product_id: str

    @validator('price_id')
    def validate_price_id(cls, v):
        if not v.startswith('price_'):
            raise ValueError('Invalid price ID format')
        return v

    @validator('product_id')
    def validate_product_id(cls, v):
        if not v.startswith('prod_'):
            raise ValueError('Invalid product ID format')
        return v

Rate Limiting

Implement rate limiting on payment endpoints:

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@router.post("/create-checkout-session")
@limiter.limit("5/minute")
async def create_checkout_session(request: Request, ...):
    # Endpoint logic
    pass

API Key Management

Never hardcode API keys in your source code:

# ❌ Bad
stripe.api_key = "sk_test_abc123..."

# ✅ Good
stripe.api_key = os.getenv("STRIPE_SECRET_KEY")

Regularly rotate your API keys:

Generate new keys in Stripe Dashboard
Update environment variables
Deploy updated configuration
Deactivate old keys

Set up monitoring for suspicious activities:

Failed webhook verifications
Unusual payment patterns
High error rates
Unauthorized access attempts

Production Deployment

Environment Configuration

Before deploying to production, ensure you're using live Stripe keys and have completed account verification.

# Production environment variables
STRIPE_SECRET_KEY=sk_live_your_live_secret_key
STRIPE_PUBLIC_KEY=pk_live_your_live_publishable_key
WEBHOOK_SECRET=whsec_your_live_webhook_secret
FRONTEND_URL=https://yourdomain.com
DATABASE_URL=postgresql://user:password@prod-db:5432/your_db

# Security settings
DEBUG=False
LOG_LEVEL=INFO
ALLOWED_HOSTS=yourdomain.com,www.yourdomain.com

Health Checks

Implement comprehensive health checks:

@router.get("/health")
async def health_check():
    """Health check endpoint for monitoring"""
    try:
        # Check database connection
        db.execute("SELECT 1")

        # Check Stripe API connection
        stripe.Account.retrieve()

        return {
            "status": "healthy",
            "timestamp": datetime.utcnow().isoformat(),
            "version": "1.0.0"
        }
    except Exception as e:
        raise HTTPException(status_code=503, detail=f"Health check failed: {str(e)}")

Monitoring and Alerts

Set up monitoring for:

Payment success/failure rates
Webhook delivery status
API response times
Database performance
Credit consumption patterns

Backup Strategy

Ensure regular backups of:

Customer payment data
Subscription information
Transaction history
Webhook event logs
Database snapshots

Consider implementing point-in-time recovery for critical payment data to minimize potential data loss.

Webhook events are rejected with signature errors
Payment updates not reflected in application

Solutions:

Verify WEBHOOK_SECRET matches Stripe dashboard
Ensure raw request body is used for verification
Check webhook endpoint URL configuration
Validate SSL certificate on webhook endpoint

Customer Not Found Errors

Symptoms:

"Customer not found" errors in logs
Billing portal access fails

Solutions:

Verify customer creation in checkout flow
Check customer ID storage in database
Ensure customer exists in Stripe dashboard
Implement customer ID validation

Subscription Status Not Updating

Symptoms:

Subscription appears active but payments failed
Credits not reset on renewal

Solutions:

Check webhook endpoint accessibility
Verify webhook event selection in Stripe
Review webhook delivery logs
Implement manual sync mechanisms

Debug Tools and Techniques

Enable Detailed Logging

import logging

# Configure detailed logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

# Log webhook events
@router.post("/webhook")
async def stripe_webhook(request: Request, ...):
    logger.info(f"Webhook received: {event['type']}")
    logger.debug(f"Event data: {event['data']}")
    # ... rest of webhook handler

Stripe Dashboard Monitoring

Use Stripe Dashboard to monitor:

Payment success rates
Webhook delivery status
Customer lifecycle events
Error patterns

Testing Webhook Delivery

# Test webhook endpoint directly
curl -X POST https://yourdomain.com/payments/webhook \
  -H "Content-Type: application/json" \
  -H "Stripe-Signature: your_test_signature" \
  -d @test_webhook_payload.json

Performance Optimization

Database Indexing

Ensure proper indexes on frequently queried columns:

-- Index for customer lookups
CREATE INDEX idx_users_customer_id ON users(customer_id);

-- Index for subscription lookups
CREATE INDEX idx_users_subscription_id ON users(subscription_id);

-- Index for plan lookups
CREATE INDEX idx_plans_product_id ON plans(product_id);

Caching Strategy

Implement caching for frequently accessed data:

from functools import lru_cache

@lru_cache(maxsize=128)
def get_plan_by_product_id(product_id: str):
    """Cached plan lookup"""
    return db.query(Plan).filter(Plan.product_id == product_id).first()

Payment System

On this page