Stripe Integration for SaaS Billing

Billing is the lifeblood of SaaS. Stripe is the payment processor used by 70% of SaaS startups, handling subscriptions, invoices, tax compliance, and payouts. This guide integrates Stripe into your FastAPI backend: creating customers, managing subscriptions, processing webhooks securely, and providing a customer billing portal. You'll handle recurring charges, plan upgrades/downgrades, and tax calculations without building a payment processor from scratch.

Why Stripe Over Rolling Your Own

Building a payment system means handling PCI-DSS compliance (payment card industry security), PCI secrets management, and dozens of edge cases (failed charges, chargebacks, tax liability). Stripe abstracts this: they store card data, handle compliance, manage failed charge retries, and provide an audit trail. You integrate via their API and focus on business logic. Stripe charges 2.9% + $0.30 per successful charge for SaaS; the saved engineering time pays for itself in weeks.

Setting Up Stripe

Install the Stripe Python library:

pip install stripe

Get your API keys from the Stripe Dashboard. Store them as environment variables:

# .env
STRIPE_SECRET_KEY=sk_live_your_live_key
STRIPE_PUBLISHABLE_KEY=pk_live_your_publishable_key
STRIPE_WEBHOOK_SECRET=whsec_your_webhook_secret

Initialize Stripe in your app:

import stripe
import os

stripe.api_key = os.getenv("STRIPE_SECRET_KEY")
STRIPE_PUBLISHABLE_KEY = os.getenv("STRIPE_PUBLISHABLE_KEY")
STRIPE_WEBHOOK_SECRET = os.getenv("STRIPE_WEBHOOK_SECRET")

Creating Stripe Customers

Map your tenants to Stripe customers:

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from pydantic import BaseModel

class Subscription(BaseModel):
    price_id: str

@app.post("/billing/create-customer")
async def create_customer(
    session: Annotated[Session, Depends(get_db)],
    current_user: Annotated[TokenData, Depends(get_current_user)]
):
    """
    Create a Stripe customer for the tenant.
    Called once during tenant onboarding.
    """
    tenant = session.query(Tenant).filter(
        Tenant.id == current_user.tenant_id
    ).first()
    
    if not tenant:
        raise HTTPException(status_code=404, detail="Tenant not found")
    
    # Check if already a Stripe customer
    if tenant.stripe_customer_id:
        return {
            "stripe_customer_id": tenant.stripe_customer_id,
            "message": "Customer already exists"
        }
    
    # Create in Stripe
    customer = stripe.Customer.create(
        email=f"billing@{tenant.slug}.example.com",
        name=tenant.name,
        metadata={"tenant_id": str(current_user.tenant_id)}
    )
    
    # Store the Stripe ID
    tenant.stripe_customer_id = customer.id
    session.commit()
    
    return {
        "stripe_customer_id": customer.id,
        "message": "Customer created successfully"
    }

Managing Subscriptions

Create a subscription when a tenant upgrades:

@app.post("/billing/subscribe")
async def create_subscription(
    request: Subscription,
    session: Annotated[Session, Depends(get_db)],
    current_user: Annotated[TokenData, Depends(get_current_user)]
):
    """
    Subscribe a tenant to a plan.
    Creates a recurring charge that will renew monthly.
    """
    tenant = session.query(Tenant).filter(
        Tenant.id == current_user.tenant_id
    ).first()
    
    if not tenant or not tenant.stripe_customer_id:
        raise HTTPException(
            status_code=400,
            detail="Tenant must create a Stripe customer first"
        )
    
    # Create subscription
    subscription = stripe.Subscription.create(
        customer=tenant.stripe_customer_id,
        items=[{"price": request.price_id}],
        payment_behavior="default_incomplete",
        expand=["latest_invoice.payment_intent"]
    )
    
    # Store subscription ID
    tenant.stripe_subscription_id = subscription.id
    tenant.subscription_status = subscription.status  # "incomplete", "active", etc.
    session.commit()
    
    return {
        "subscription_id": subscription.id,
        "status": subscription.status,
        "client_secret": subscription.latest_invoice.payment_intent.client_secret
    }

The client_secret is passed to the frontend's Stripe.js to complete payment via 3D Secure (SCA) if required.

Handling Subscription Changes

Upgrade or downgrade a subscription:

@app.post("/billing/change-plan")
async def change_subscription_plan(
    new_price_id: str,
    session: Annotated[Session, Depends(get_db)],
    current_user: Annotated[TokenData, Depends(get_current_user)]
):
    """
    Change the subscription plan.
    Prorates charges: if upgrading mid-cycle, charges the difference; if downgrading, credits.
    """
    tenant = session.query(Tenant).filter(
        Tenant.id == current_user.tenant_id
    ).first()
    
    if not tenant or not tenant.stripe_subscription_id:
        raise HTTPException(status_code=400, detail="No active subscription")
    
    # Get current subscription
    subscription = stripe.Subscription.retrieve(tenant.stripe_subscription_id)
    
    # Update the price
    updated_subscription = stripe.Subscription.modify(
        tenant.stripe_subscription_id,
        items=[{
            "id": subscription.items.data[0].id,
            "price": new_price_id
        }],
        proration_behavior="create_prorations"  # Prorate charges
    )
    
    tenant.subscription_status = updated_subscription.status
    session.commit()
    
    return {
        "subscription_id": updated_subscription.id,
        "status": updated_subscription.status,
        "new_amount": updated_subscription.items.data[0].price.unit_amount / 100  # In dollars
    }

Processing Webhooks Securely

Stripe sends webhooks for events: payment succeeded, subscription renewed, failed charge, etc. Webhooks are critical for billing accuracy (don't trust client-side assertions about payment):

from fastapi import Request, HTTPException
import hmac
import hashlib

@app.post("/billing/webhooks/stripe")
async def stripe_webhook(
    request: Request,
    session: Annotated[Session, Depends(get_db)]
):
    """
    Webhook endpoint for Stripe events.
    Verifies the signature to ensure the request is authentic.
    """
    # Read raw body (required for signature verification)
    body = await request.body()
    signature = request.headers.get("stripe-signature")
    
    # Verify signature
    try:
        event = stripe.Webhook.construct_event(
            body,
            signature,
            STRIPE_WEBHOOK_SECRET
        )
    except ValueError:
        raise HTTPException(status_code=400, detail="Invalid payload")
    except stripe.error.SignatureVerificationError:
        raise HTTPException(status_code=400, detail="Invalid signature")
    
    # Handle specific events
    if event["type"] == "customer.subscription.updated":
        handle_subscription_updated(event["data"]["object"], session)
    
    elif event["type"] == "customer.subscription.deleted":
        handle_subscription_deleted(event["data"]["object"], session)
    
    elif event["type"] == "invoice.paid":
        handle_invoice_paid(event["data"]["object"], session)
    
    elif event["type"] == "invoice.payment_failed":
        handle_invoice_payment_failed(event["data"]["object"], session)
    
    return {"received": True}

def handle_subscription_updated(subscription_data: dict, session: Session):
    """Handle subscription updates."""
    tenant = session.query(Tenant).filter(
        Tenant.stripe_subscription_id == subscription_data["id"]
    ).first()
    
    if tenant:
        tenant.subscription_status = subscription_data["status"]
        tenant.subscription_current_period_end = datetime.fromtimestamp(
            subscription_data["current_period_end"]
        )
        session.commit()
        print(f"Updated subscription for tenant {tenant.id}: {subscription_data['status']}")

def handle_invoice_paid(invoice_data: dict, session: Session):
    """Log successful payment."""
    print(f"Invoice {invoice_data['id']} paid: ${invoice_data['total'] / 100}")
    # Log to analytics, send receipt email, etc.

def handle_invoice_payment_failed(invoice_data: dict, session: Session):
    """Handle failed payment (dunning)."""
    tenant = session.query(Tenant).filter(
        Tenant.stripe_customer_id == invoice_data["customer"]
    ).first()
    
    if tenant:
        # Send email to tenant: "Your payment failed. Update your card."
        # Stripe retries automatically; you can also implement custom dunning logic
        print(f"Payment failed for tenant {tenant.id}")

Billing Portal

Let tenants manage their subscription in Stripe's hosted portal:

@app.post("/billing/portal")
async def create_billing_portal_session(
    session: Annotated[Session, Depends(get_db)],
    current_user: Annotated[TokenData, Depends(get_current_user)]
):
    """
    Create a session for Stripe's billing portal.
    User redirects to the returned URL to manage payment method, view invoices, etc.
    """
    tenant = session.query(Tenant).filter(
        Tenant.id == current_user.tenant_id
    ).first()
    
    if not tenant or not tenant.stripe_customer_id:
        raise HTTPException(status_code=400, detail="No Stripe customer")
    
    portal_session = stripe.BillingPortal.Session.create(
        customer=tenant.stripe_customer_id,
        return_url="https://yourapp.com/settings/billing"
    )
    
    return {"url": portal_session.url}

The user visits this URL to:

• Update payment method
• Download invoices
• Cancel subscription
• Change billing email

Tax Compliance

Stripe Tax calculates sales tax automatically. When creating a subscription, pass the customer's tax ID:

@app.post("/billing/subscribe-with-tax")
async def create_subscription_with_tax(
    request: Subscription,
    session: Annotated[Session, Depends(get_db)],
    current_user: Annotated[TokenData, Depends(get_current_user)]
):
    """
    Create subscription with automatic tax calculation.
    Stripe Tax determines applicable rates by customer location.
    """
    tenant = session.query(Tenant).filter(
        Tenant.id == current_user.tenant_id
    ).first()
    
    # Enable automatic tax
    subscription = stripe.Subscription.create(
        customer=tenant.stripe_customer_id,
        items=[{"price": request.price_id}],
        automatic_tax={"enabled": True},
        billing_cycle_anchor=int(datetime.utcnow().timestamp())
    )
    
    return {"subscription_id": subscription.id}

Stripe calculates and collects tax automatically; you remit to authorities via Stripe Tax Dashboard.

Key Takeaways

• Create a Stripe customer for each tenant during onboarding; store the customer ID in your database.
• Subscriptions are created with a price ID; Stripe handles recurring charges automatically.
• Webhooks are the source of truth for billing state (payment success, renewal, failure). Verify signatures.
• Offer a billing portal (Stripe-hosted) for tenants to manage payment methods and view invoices.
• Enable Stripe Tax for automatic calculation; Stripe remits to authorities on your behalf.

Frequently Asked Questions

What if a customer's card is declined?

Stripe retries automatically (up to 3–5 times with backoff). If all retries fail, Stripe sends a invoice.payment_failed webhook. You can implement dunning (email → payment retry → grace period → cancel) in the webhook handler.

How do I offer a free trial?

Create a plan with a trial_period_days parameter:

stripe.Subscription.create(
    customer=stripe_customer_id,
    items=[{"price": price_id}],
    trial_period_days=14
)

During trial, subscription.status = "trialing". After trial, Stripe charges automatically.

Can I prorate when a customer cancels mid-cycle?

Yes. Stripe handles proration automatically. Use stripe.Subscription.delete(subscription_id, prorate=True) to issue a final credit.

How do I handle refunds?

Use stripe.Refund.create(charge=charge_id, amount=1000) (amount in cents). Stripe can refund within 180 days. For subscriptions, refund the last invoice: stripe.Invoice.retrieve(invoice_id).paid → refund if paid.

Should I store sensitive payment data?

Never. Stripe tokenizes cards; store the token ID, not the card number. Stripe.js handles card collection on the frontend. Your backend never sees raw card data.

Further Reading

• Stripe API Documentation
• Stripe Billing Guide
• Stripe Webhooks Security
• Stripe Tax Integration