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_subscri...
Source：https://dev.to/patrykkomow12/stripe-subscriptions-in-fastapi-with-async-sqlalchemy-4lj9
My favorites