Stripe subscriptions in FastAPI with async SQLAlchemy

Checkout, a local copy of subscriptions in PostgreSQL, webhooks that survive duplicates and reordering, and the customer portal. This is the setup I use in production, with the parts that cost me time spelled out.

The shape of it

Stripe is the source of truth for who pays for what. The app keeps a local copy of the parts it needs (customers, subscriptions, handled events) so that a page can check a user's plan without an API call. Three things write to that copy:

1. Starting Checkout creates the Stripe customer and remembers it.
2. The success page, where Stripe sends the user after paying, syncs the new subscription at once so the user sees their plan immediately.
3. Webhooks keep everything in sync afterwards: renewals, failed payments, cancellations, plan changes made in the customer portal.

The stack: FastAPI, SQLAlchemy 2.0 in async mode with asyncpg, PostgreSQL and the official stripe package. The code below is trimmed from a working app; imports are left out where they're obvious.

One module for every Stripe call

The newer StripeClient has *_async versions of every method, which is what you want inside an async app. Two details matter. First, give it an HTTPX client so those calls really are non-blocking. Second, the objects it returns look like dicts but aren't, so convert them once, at the boundary:

# app/billing/stripe_gateway.py
from functools import lru_cache

import stripe

from app.config import get_settings


@lru_cache
def client() -> stripe.StripeClient:
    return stripe.StripeClient(
        get_settings().stripe_secret_key, http_client=stripe.HTTPXClient()
    )


async def create_customer(email: str, user_id: int) -> str:
    customer = await client().v1.customers.create_async(
        params={"email": email, "metadata": {"user_id": str(user_id)}}
    )
    return customer.id


async def create_checkout_session(params: dict) -> dict:
    return (await client().v1.checkout.sessions.create_async(params=params)).to_dict()


async def retrieve_checkout_session(session_id: str) -> dict:
    return (await client().v1.checkout.sessions.retrieve_async(session_id)).to_dict()


async def retrieve_subscription(subscription_id: str) -> dict:
    return (await client().v1.subscriptions.retrieve_async(subscription_id)).to_dict()


def construct_event(payload: bytes, signature: str | None) -> dict:
    """Checks the webhook signature. Raises ValueError or
    stripe.SignatureVerificationError for anything Stripe didn't send."""
    event = client().construct_event(
        payload, signature, get_settings().stripe_webhook_secret
    )
    return event.to_dict()

Every other module imports this one as gateway and never touches the SDK. The payoff comes in tests: replace these few functions with a fake and no test needs the network, a recorded cassette or a mock per call site.

Three tables

Plans live in configuration, not in the database. A setting maps your plan names to Stripe price ids:

STRIPE_PLANS={"starter": "price_1Q...", "pro": "price_1Q..."}

The tables hold the link between a user and their Stripe customer, the subscriptions, and the ids of webhook events already handled:

# app/billing/models.py
# Statuses that give access. past_due keeps access while Stripe retries a failed payment.
ACCESS_STATUSES = ("active", "trialing", "past_due")


class BillingCustomer(Base):
    __tablename__ = "billing_customers"

    user_id: Mapped[int] = mapped_column(
        ForeignKey("users.id", ondelete="CASCADE"), primary_key=True
    )
    stripe_customer_id: Mapped[str] = mapped_column(String(255), unique=True)


class Subscription(Base):
    __tablename__ = "subscriptions"

    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(
        ForeignKey("users.id", ondelete="CASCADE"), index=True
    )
    stripe_subscription_id: Mapped[str] = mapped_column(String(255), unique=True)
    plan: Mapped[str | None] = mapped_column(String(50))  # key from STRIPE_PLANS
    price_id: Mapped[str] = mapped_column(String(255))
    status: Mapped[str] = mapped_column(String(30))  # Stripe's status, as is
    current_period_end: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    cancel_at_period_end: Mapped[bool] = mapped_column(default=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )


class StripeEvent(Base):
    """Webhook events already handled. Stripe can deliver an event more than once."""

    __tablename__ = "stripe_events"

    id: Mapped[str] = mapped_column(String(255), primary_key=True)
    type: Mapped[str] = mapped_column(String(100))

The unique constraints on stripe_customer_id and stripe_subscription_id aren't decoration. They are what makes the next sections safe.

Starting Checkout

Create the Stripe customer yourself before Checkout, instead of letting Checkout create one. Then every subscription, invoice and portal session for a user hangs off the same customer, and you can look users up by customer id in webhooks.

# app/billing/service.py
from sqlalchemy.dialects.postgresql import insert as pg_insert


async def _insert_ignore(session: AsyncSession, model: type, **values) -> None:
    """INSERT ... ON CONFLICT DO NOTHING."""
    await session.execute(pg_insert(model).values(**values).on_conflict_do_nothing())


async def get_customer_id(session: AsyncSession, user: User) -> str | None:
    customer = await session.get(BillingCustomer, user.id)
    return customer.stripe_customer_id if customer else None


async def ensure_customer(session: AsyncSession, user: User) -> str:
    if customer_id := await get_customer_id(session, user):
        return customer_id
    customer_id = await gateway.create_customer(user.email, user.id)
    await _insert_ignore(
        session, BillingCustomer, user_id=user.id, stripe_customer_id=customer_id
    )
    await session.commit()
    # On a double click both requests create a customer; the first one saved wins.
    return await get_customer_id(session, user) or customer_id

Then the Checkout parameters. Put the user id in client_reference_id and in metadata: the success page uses the first to check the session belongs to the current user, and the metadata travels with the subscription for debugging in the Dashboard.

def checkout_params(user: User, plan: Plan, customer_id: str) -> dict:
    base_url = get_settings().base_url
    metadata = {"user_id": str(user.id), "plan": plan.key}
    return {
        "mode": "subscription",
        "customer": customer_id,
        "client_reference_id": str(user.id),
        "line_items": [{"price": plan.price_id, "quantity": 1}],
        "success_url": base_url + "/billing/success?session_id={CHECKOUT_SESSION_ID}",
        "cancel_url": f"{base_url}/pricing",
        "allow_promotion_codes": True,
        "metadata": metadata,
        "subscription_data": {"metadata": metadata},
    }

Note that the success URL is a plain string, not an f-string: {CHECKOUT_SESSION_ID} is a placeholder Stripe fills in, so Python must leave it alone. Also note what's missing: payment_method_types. Leave it out and Checkout offers whatever payment methods you switch on in the Dashboard, filtered by currency and by what works for subscriptions. Hardcoding ["card"] means a code change every time you want to offer something new.

The route is short. Redirect with 303 so the browser follows with a GET:

@router.post("/billing/checkout/{plan_key}")
async def checkout(plan_key: str, session: SessionDep, user: VerifiedUser):
    plan = await service.get_plan(plan_key)
    if plan is None:
        raise HTTPException(status_code=404)
    access = await service.get_access(session, user)
    if access.subscription:
        # Plan changes and cancellations happen in the customer portal.
        return RedirectResponse("/billing", status_code=303)

    customer_id = await service.ensure_customer(session, user)
    checkout = await gateway.create_checkout_session(
        service.checkout_params(user, plan, customer_id)
    )
    return RedirectResponse(checkout["url"], status_code=303)

The check for an existing subscription matters more than it looks. Without it, a user who opens the pricing page in two tabs can end up with two subscriptions and two charges every month.

The success page and the race with the webhook

If you only rely on webhooks, the user lands on your success page, sees "Free plan", and refreshes in a panic a second before the webhook arrives. So the success page syncs too:

@router.get("/billing/success")
async def checkout_success(session: SessionDep, user: CurrentUser, session_id: str = ""):
    try:
        checkout = await gateway.retrieve_checkout_session(session_id)
    except stripe.InvalidRequestError:
        raise HTTPException(status_code=404) from None
    if checkout.get("client_reference_id") != str(user.id):
        raise HTTPException(status_code=404)

    await service.sync_checkout_session(session, checkout)
    await session.commit()
    return RedirectResponse("/billing", status_code=303)

Never trust session_id from the query string on its own: fetch the session from Stripe and check it belongs to the signed-in user.

Now the race. Stripe redirects the user and fires checkout.session.completed at practically the same moment. Both handlers try to insert the same subscription; one of them hits the unique constraint and the user gets a 500 on the page that should say "thanks". The fix is to make both paths idempotent: insert with ON CONFLICT DO NOTHING, then update whatever row is there.

async def sync_subscription(session: AsyncSession, data: dict) -> Subscription | None:
    """Creates or updates the local copy of a Stripe subscription."""
    user_id = await session.scalar(
        select(BillingCustomer.user_id)
        .where(BillingCustomer.stripe_customer_id == data["customer"])
    )
    if user_id is None:
        logger.warning("Subscription %s belongs to no known user", data["id"])
        return None

    item = data["items"]["data"][0]
    price_id = item["price"]["id"]
    await _insert_ignore(
        session,
        Subscription,
        user_id=user_id,
        stripe_subscription_id=data["id"],
        price_id=price_id,
        status=data["status"],
    )
    subscription = await session.scalar(
        select(Subscription).where(Subscription.stripe_subscription_id == data["id"])
    )
    subscription.price_id = price_id
    subscription.plan = plan_key_for_price(price_id)
    subscription.status = data["status"]
    # Since Stripe API 2025-03-31 the billing period lives on the subscription item.
    subscription.current_period_end = _timestamp(item.get("current_period_end"))
    subscription.cancel_at_period_end = bool(
        data.get("cancel_at_period_end") or data.get("cancel_at")
    )
    await session.flush()
    return subscription


async def sync_checkout_session(session: AsyncSession, checkout: dict) -> None:
    if checkout["mode"] == "subscription" and checkout.get("subscription"):
        data = await gateway.retrieve_subscription(checkout["subscription"])
        await sync_subscription(session, data)

Two things in there come from newer Stripe API versions and are easy to miss if you learned Stripe from older tutorials:

• current_period_end moved from the subscription to its items in API version 2025-03-31. Read it from items.data[0].
• Cancelling in the customer portal can set cancel_at (a timestamp) and leave cancel_at_period_end false. If you only check the boolean, a cancelled subscription looks like it renews. Check both.

The webhook endpoint

Three rules: read the raw body (the signature is computed over the exact bytes, so parsing JSON first breaks it), verify the signature, and answer 2xx quickly once you've handled the event.

# No CSRF dependency here: Stripe's signature protects this route instead.
webhook_router = APIRouter()


@webhook_router.post("/stripe/webhook")
async def stripe_webhook(request: Request, session: SessionDep):
    payload = await request.body()
    try:
        event = gateway.construct_event(payload, request.headers.get("stripe-signature"))
    except (ValueError, stripe.SignatureVerificationError):
        logger.warning("Rejected a Stripe webhook with an invalid signature")
        return Response(status_code=400)

    handled = await service.handle_event(session, event)
    logger.info("Stripe event %s %s", event["type"], "handled" if handled else "skipped")
    return Response(status_code=200)

If your forms use a CSRF dependency on the whole router, keep the webhook on a separate router without it, or every Stripe delivery fails with 403.

A 400 from this endpoint almost always means the signing secret doesn't belong to this endpoint. Each endpoint has its own whsec_..., and so do test and live mode, and so does stripe listen on your machine.

Applying events: duplicates and order

Stripe guarantees "at least once", not "exactly once", and not in order. Two consequences:

1. Duplicates. Store each handled event.id and skip ones you've seen. Save the id in the same transaction as the changes, so an event that fails halfway isn't marked as handled and Stripe's retry runs it again.
2. Order. customer.subscription.updated can arrive before customer.subscription.created. Don't apply the payload as "the latest state". Fetch the subscription from Stripe and save that.

SUBSCRIPTION_EVENTS = {
    "customer.subscription.created",
    "customer.subscription.updated",
    "customer.subscription.deleted",
    "customer.subscription.paused",
    "customer.subscription.resumed",
}


async def handle_event(session: AsyncSession, event: dict) -> bool:
    """Applies one webhook event. Returns False if it was already handled."""
    if await session.get(StripeEvent, event["id"]):
        return False

    event_type = event["type"]
    obj = event["data"]["object"]

    if event_type == "checkout.session.completed":
        await sync_checkout_session(session, obj)

    elif event_type in SUBSCRIPTION_EVENTS:
        # Events can arrive out of order, so always apply Stripe's current state.
        data = await gateway.retrieve_subscription(obj["id"])
        await sync_subscription(session, data)

    elif event_type in ("invoice.paid", "invoice.payment_failed"):
        # A renewal or a failed renewal changes the subscription's status and period.
        if subscription_id := _subscription_id_from_invoice(obj):
            await sync_subscription(
                session, await gateway.retrieve_subscription(subscription_id)
            )

    session.add(StripeEvent(id=event["id"], type=event_type))
    try:
        await session.commit()
    except IntegrityError:
        # The same event was handled by a parallel request a moment ago.
        await session.rollback()
        return False
    return True


def _subscription_id_from_invoice(invoice: dict) -> str | None:
    # API 2025-03-31+: invoice.parent.subscription_details.subscription
    parent = invoice.get("parent") or {}
    details = parent.get("subscription_details") or {}
    return details.get("subscription") or invoice.get("subscription")

The IntegrityError branch handles the case the session.get check can't: two deliveries of the same event processed at the same time. Both pass the check, both do the work (which is harmless, because every write is an upsert of Stripe's current state), and the second commit fails on the primary key and is rolled back.

Fetching the subscription on every event costs one API call per event. For a SaaS that's nothing, and it removes a whole class of "the status went back to incomplete" bugs.

In the Dashboard, subscribe the endpoint to the events you handle. For subscriptions that's:

checkout.session.completed
customer.subscription.created
customer.subscription.updated
customer.subscription.deleted
customer.subscription.paused
customer.subscription.resumed
invoice.paid
invoice.payment_failed

Checking access in your routes

With the local copy in place, a paid feature checks the database, not Stripe:

async def get_plan_for(session: AsyncSession, user: User) -> str | None:
    subscription = await session.scalar(
        select(Subscription)
        .where(
            Subscription.user_id == user.id,
            Subscription.status.in_(ACCESS_STATUSES),
        )
        .order_by(Subscription.created_at.desc())
        .limit(1)
    )
    return subscription.plan if subscription else None

past_due is in ACCESS_STATUSES on purpose. When a renewal fails, Stripe retries for days (you set how long in Settings → Billing → Subscriptions and emails). Locking the user out on the first failed attempt punishes someone whose card just expired. When Stripe gives up, the subscription becomes canceled or unpaid, a webhook arrives, and access ends.

A cancelled subscription stays active until the end of the paid period, with cancel_at_period_end set, so the user keeps what they paid for. Show "Your plan ends on …" from current_period_end instead of hiding the cancellation.

Cancelling and changing plans: the customer portal

Don't build plan changes, card updates and cancellation yourself. Stripe's customer portal does all of it, with proration, and every change comes back to you as a customer.subscription.updated webhook that the code above already handles.

@router.post("/billing/portal")
async def customer_portal(session: SessionDep, user: CurrentUser):
    customer_id = await service.get_customer_id(session, user)
    if customer_id is None:
        return RedirectResponse("/pricing", status_code=303)
    url = await gateway.create_portal_session(customer_id, f"{get_settings().base_url}/billing")
    return RedirectResponse(url, status_code=303)

In the gateway, create_portal_session is one more thin wrapper: it calls client().v1.billing_portal.sessions.create_async() with the customer and return URL and returns the session's url. Configure what the portal allows (switching plans, cancelling now or at period end, invoice history) in Settings → Billing → Customer portal, or create a portal configuration through the API and pass its id as configuration.

Testing it locally

The Stripe CLI forwards test-mode events to your machine and prints a signing secret for it:

stripe listen --forward-to localhost:8000/stripe/webhook
# Ready! Your webhook signing secret is whsec_...  (put it in STRIPE_WEBHOOK_SECRET)

Then go through the flow by hand, because that's where the race and the ordering bugs show up:

1. Subscribe with the test card 4242 4242 4242 4242, any future date and CVC. The plan should show up on the page Stripe sends you back to, before you refresh.
2. Open the customer portal and cancel. Your billing page should show the end date, and the subscription should still be active.
3. Pay with 4000 0000 0000 0002 to see a declined card in Checkout. For a failed renewal, send the event by hand with stripe trigger invoice.payment_failed (it creates its own test customer, so your handler should log it and move on rather than crash).
4. Resend an event from the Dashboard (or run the same stripe events resend evt_... twice) and check that nothing changes the second time.

In tests, replace the gateway functions with a fake that keeps Stripe objects in a dict, and pass event dicts straight to handle_event. Then the success-page race, duplicate events and out-of-order events each become an ordinary pytest test that runs in milliseconds.

Checklist

• All Stripe calls in one module, returning plain dicts.
• Create the customer yourself; one customer per user, enforced by a unique constraint.
• Upserts (ON CONFLICT DO NOTHING + update) wherever the success page and webhooks meet.
• Webhook: raw body, signature check, no CSRF, event id saved in the same transaction.
• On subscription events, fetch the current subscription instead of trusting the payload.
• Read current_period_end from the item and check both cancel_at_period_end and cancel_at.
• Let the customer portal handle changes and cancellations.

---

This is the billing module from getpylaunch, a FastAPI SaaS starter I sell, where it comes with EU VAT, invoices and the tests already written.