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. Built with async/await patterns for maximum performance.

Architecture

Async-First Design

All payment endpoints and service functions are fully asynchronous:

# All payment endpoints use async/await
@router.get("/get-user-subscription")
async def get_user_subscription(user: user_dependency, db: db_dependency):
    return await get_user_subscription_details(user, db)

@router.post("/create-checkout-session")
async def create_checkout_session(...):
    return await create_subscription_checkout_session(...)

Dual Database Session Pattern

The system uses two types of database sessions:

Understanding the dual session pattern is critical for correctly implementing payment features and background tasks.

AsyncSession - For FastAPI endpoints:

from app.db.database import db_dependency, AsyncSessionLocal
from sqlalchemy.ext.asyncio import AsyncSession

# FastAPI endpoints use async sessions
@router.get("/payments/example")
async def example_endpoint(db: db_dependency):  # db is AsyncSession
    result = await db.execute(select(User).filter(User.id == user_id))
    user = result.scalar_one_or_none()
    await db.commit()

SyncSession - For Celery background tasks:

from app.db.database import SyncSessionLocal

# Celery tasks use sync sessions
@celery_app.task()
def process_payment_task(user_id: int):
    db = SyncSessionLocal()
    try:
        user = db.query(User).filter(User.id == user_id).first()
        db.commit()
    finally:
        db.close()

Never mix async and sync sessions! Using AsyncSession in Celery tasks or SyncSession in FastAPI endpoints will cause runtime errors.

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+asyncpg://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[asyncio] - Async database ORM
asyncpg - Async PostgreSQL driver
psycopg2-binary - Sync PostgreSQL driver (for Celery)
alembic - Database migrations
pydantic~=2.9.2 - 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
) VALUES (
    'Pro Plan',
    'prod_abc123',
    'price_def456',
    '29.99',
    'Professional Plan',
    'Pro Monthly'
);

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

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:

from sqlalchemy import Column, Integer, String
from app.db.database import Base

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)

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)

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)

All payment-related database operations use async SQLAlchemy 2.0 syntax with select(), execute(), and await patterns.

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"
}

Implementation:

@router.post("/create-checkout-session")
async def create_checkout_session(
        request: Request,
        user: user_dependency,
        create_checkout_request: CreateCheckoutRequest,
        db: db_dependency
):
    """Create a checkout session for subscription payments"""
    if user.subscription_status == "active":
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="You already have an ongoing subscription. Cancel it first."
        )

    try:
        return await create_subscription_checkout_session(
            user,
            create_checkout_request.price_id,
            FRONTEND_URL,
            db
        )
    except stripe.error.StripeError as e:
        logger.error(f"Error creating checkout session: {str(e)}")
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"Stripe error: {str(e)}"
        )

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"
}

Implementation:

@router.post("/create-checkout-session-onetime")
async def create_checkout_session_onetime(
        request: Request,
        user: user_dependency,
        create_checkout_request: CreateCheckoutRequest,
        db: db_dependency
):
    """Create a checkout session for one-time payments"""
    try:
        return await create_onetime_checkout_session(
            user,
            create_checkout_request.price_id,
            create_checkout_request.product_id,
            FRONTEND_URL,
            db
        )
    except Exception as e:
        logger.error(f"Error creating one-time checkout session: {str(e)}")
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Error creating checkout session"
        )

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.

Implementation:

@router.get("/get-user-subscription")
async def get_user_subscription(user: user_dependency, db: db_dependency):
    """Get user's current subscription details"""
    try:
        return await get_user_subscription_details(user, db)
    except stripe.error.StripeError as e:
        logger.error(f"Error retrieving subscription: {str(e)}")
        raise HTTPException(status_code=400, detail="Error retrieving subscription")

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.

Service Layer Architecture

Async Service Functions

All payment service functions are fully asynchronous:

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

async def get_user_by_subscription_id(
    subscription_id: str,
    db: AsyncSession
) -> Optional[User]:
    """Get user by subscription ID"""
    stmt = select(User).filter(User.subscription_id == subscription_id)
    result = await db.execute(stmt)
    return result.scalar_one_or_none()

async def create_or_get_customer(user: User, db: AsyncSession) -> str:
    """Create a Stripe customer if one doesn't exist"""
    if not user.customer_id:
        stripe_customer = stripe.Customer.create(
            email=user.email,
            metadata={"user_id": user.id}
        )
        user.customer_id = stripe_customer.id
        await db.commit()

    return user.customer_id

async def create_subscription_checkout_session(
        user: User,
        price_id: str,
        frontend_url: str,
        db: AsyncSession
) -> str:
    """Create a checkout session for subscription payments"""
    stripe_customer_id = await create_or_get_customer(user, db)

    checkout_session = stripe.checkout.Session.create(
        customer=stripe_customer_id,
        client_reference_id=user.id,
        success_url=f"{frontend_url}/payments/success?session_id={{CHECKOUT_SESSION_ID}}",
        cancel_url=f"{frontend_url}/payments/cancel",
        payment_method_types=["card"],
        mode="subscription",
        subscription_data={
            "trial_period_days": 14,
        },
        line_items=[{"price": price_id, "quantity": 1}]
    )
    return checkout_session.url

Always use await when calling async service functions. Forgetting await will return a coroutine object instead of the actual result.

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 with async functions:

Checkout Session Completed

Processes new subscriptions and one-time purchases:

async def handle_checkout_completed(session: Dict[str, Any], db: AsyncSession):
    """Handle successful checkout completion"""
    if session["mode"] == "subscription":
        # Get customer from Stripe
        customer_id = session.get("customer")
        user = await get_user_by_customer_id(customer_id, db)

        if user:
            # Update subscription details
            subscription = stripe.Subscription.retrieve(session["subscription"])
            user.subscription_id = subscription.id
            user.subscription_status = subscription.status

            await db.commit()

    elif session["mode"] == "payment":
        # Process one-time purchase
        user = await get_user_by_id(session["client_reference_id"], db)

        if user:
            # Add credits to user account
            # Implementation depends on your credit system
            await db.commit()

Invoice Paid

Handles subscription renewals:

async def handle_invoice_paid(session: Dict[str, Any], db: AsyncSession):
    """Handle successful invoice payment"""
    subscription_id = session.get("subscription")

    if subscription_id:
        user = await get_user_by_subscription_id(subscription_id, db)

        if user:
            # Update subscription status
            user.subscription_status = "active"
            user.subscription_last_renew = format_timestamp(session["period_start"])
            user.subscription_next_renew = format_timestamp(session["period_end"])

            await db.commit()

Payment Failed

Manages failed payments:

async def handle_invoice_payment_failed(session: Dict[str, Any], db: AsyncSession):
    """Handle failed payment"""
    subscription_id = session.get("subscription")

    if subscription_id:
        user = await get_user_by_subscription_id(subscription_id, db)

        if user:
            # Update subscription status
            user.subscription_status = "past_due"
            await db.commit()

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);
};

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

import pytest
from httpx import AsyncClient
from app.main import app

@pytest.mark.asyncio
async def test_create_subscription_checkout():
    """Test subscription checkout creation"""
    async with AsyncClient(app=app, base_url="http://test") as client:
        # Setup test user and auth
        user = await create_test_user()
        token = get_test_token(user)

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

        assert response.status_code == 200
        data = response.json()
        assert "checkout_url" in data or isinstance(data, str)

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

@pytest.mark.asyncio
async def test_get_user_subscription():
    """Test retrieving user subscription"""
    async with AsyncClient(app=app, base_url="http://test") as client:
        user = await create_test_user_with_subscription()
        token = get_test_token(user)

        response = await client.get(
            "/payments/get-user-subscription",
            headers={"Authorization": f"Bearer {token}"}
        )

        assert response.status_code == 200
        data = response.json()
        assert "subscription" in data

All tests must use async test functions with @pytest.mark.asyncio decorator when testing async endpoints.

Security Considerations

Webhook Security

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

@router.post("/webhook")
async def stripe_webhook(
    db: db_dependency,
    request: Request,
    stripe_signature: str = Header(None)
):
    """Handle Stripe webhook events"""
    raw_body = await request.body()

    # Verify webhook signature
    try:
        event = stripe.Webhook.construct_event(
            raw_body,
            stripe_signature,
            WEBHOOK_SECRET
        )
    except ValueError:
        logger.error("Invalid webhook payload")
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Invalid payload"
        )
    except stripe.error.SignatureVerificationError:
        logger.error("Invalid webhook signature")
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Invalid signature"
        )

    # Process event
    session = event["data"]["object"]

    try:
        if event["type"] == "checkout.session.completed":
            await handle_checkout_completed(session, db)
        elif event["type"] == "invoice.paid":
            await handle_invoice_paid(session, db)
        elif event["type"] == "invoice.payment_failed":
            await handle_invoice_payment_failed(session, db)

        return {"status": "success"}
    except Exception as e:
        logger.error(f"Error processing webhook: {str(e)}")
        raise HTTPException(status_code=500, detail="Error processing webhook")

Input Validation

Use Pydantic 2.x models for robust request validation:

from pydantic import BaseModel, field_validator

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

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

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

This codebase uses Pydantic 2.x. Use @field_validator instead of @validator and model_dump() instead of dict().

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
import os
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+asyncpg://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(db: db_dependency):
    """Health check endpoint for monitoring"""
    try:
        # Check database connection
        await 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
Async task completion rates

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.

Troubleshooting

Common Issues and Solutions

Webhook Signature Verification Fails

Symptoms:

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

Coroutine Never Awaited Warnings

Symptoms:

RuntimeWarning: coroutine was never awaited
Endpoints return coroutine objects instead of data

Solutions:

Ensure all async functions are called with await
Verify endpoint is defined as async def
Check service functions are properly awaited
Review async/await patterns in code

Common Mistake: Forgetting await when calling async service functions will cause the function to return a coroutine object instead of the result. Always use await with async functions!

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);

Async Query Optimization

Use proper async query patterns:

from sqlalchemy import select
from sqlalchemy.orm import defer

# Efficient async query with deferred loading
async def get_user_for_payment(user_id: int, db: AsyncSession):
    """Get user with optimized loading for payment operations"""
    stmt = select(User).filter(User.id == user_id).options(
        defer(User.hashed_password),  # Don't load sensitive data
        defer(User.google_sub)        # Don't load OAuth data
    )
    result = await db.execute(stmt)
    return result.scalar_one_or_none()

Best Practices

Async/Await Patterns

Critical: Always maintain proper async patterns throughout your payment implementation.

✅ Correct Async Patterns:

# Correct: Async endpoint with await
@router.post("/payments/example")
async def example_endpoint(db: db_dependency):
    result = await async_service_function(db)
    return result

# Correct: Async service function
async def async_service_function(db: AsyncSession):
    stmt = select(User).filter(User.id == 1)
    result = await db.execute(stmt)
    user = result.scalar_one_or_none()
    await db.commit()
    return user

❌ Common Mistakes:

# Wrong: Sync function calling async without await
def sync_endpoint(db: db_dependency):
    result = async_service_function(db)  # ❌ Missing await
    return result

# Wrong: Async endpoint not awaiting service call
async def async_endpoint(db: db_dependency):
    result = async_service_function(db)  # ❌ Missing await
    return result

# Wrong: Using sync session in async context
async def wrong_pattern():
    db = SyncSessionLocal()  # ❌ Wrong session type
    user = db.query(User).first()  # ❌ Sync query in async function

Database Session Management

For FastAPI Endpoints (AsyncSession):

@router.get("/example")
async def example(db: db_dependency):
    # db is automatically an AsyncSession via dependency injection
    stmt = select(User).filter(User.id == 1)
    result = await db.execute(stmt)
    user = result.scalar_one_or_none()
    # No need to close - handled by dependency
    return user

For Celery Tasks (SyncSession):

from app.db.database import SyncSessionLocal

@celery_app.task()
def background_payment_task(user_id: int):
    db = SyncSessionLocal()
    try:
        user = db.query(User).filter(User.id == user_id).first()
        # Perform operations
        db.commit()
    finally:
        db.close()  # Always close sync session

Error Handling

@router.post("/payments/example")
async def example(db: db_dependency):
    try:
        # Your payment logic
        result = await process_payment(db)
        return result
    except stripe.error.CardError as e:
        # Card was declined
        logger.error(f"Card error: {str(e)}")
        raise HTTPException(status_code=402, detail=str(e))
    except stripe.error.StripeError as e:
        # Other Stripe errors
        logger.error(f"Stripe error: {str(e)}")
        raise HTTPException(status_code=400, detail="Payment processing error")
    except Exception as e:
        # Unexpected errors
        logger.error(f"Unexpected error: {str(e)}")
        raise HTTPException(status_code=500, detail="Internal server error")

✅ Async-first architecture with proper async/await patterns ✅ Dual session support for FastAPI (AsyncSession) and Celery (SyncSession) ✅ Pydantic 2.x compatibility with modern validation patterns ✅ Production-ready with comprehensive error handling ✅ Secure webhook integration with signature verification ✅ Complete Stripe integration for subscriptions and one-time payments ✅ Test-ready with proper async test patterns

By following these patterns and best practices, you'll have a robust, scalable payment system that can handle production workloads efficiently.

Payment System

On this page