Integrating Stripe with Python APIs: A Step-by-Step Guide for Micro-SaaS Builders
Payment code is the shortest path between a working Python API and a business, and it is also the one subsystem where a silent bug costs you money on every request rather than just an error log line. This guide walks the full integration against FastAPI: pinned SDK initialisation, a plan catalogue your code can trust, dynamic checkout sessions, signature-verified webhooks, and the subscription state machine that decides whether a given API key still works this morning. Part of the Building & Monetizing API-Driven Micro-SaaS guide.
The mental model that matters is this: Stripe owns the money, your database owns the entitlement, and exactly one piece of code — the webhook handler — is allowed to move information between them. Builders who let the checkout success redirect grant access end up with customers who bookmarked a URL and get a free tier upgrade forever. Builders who poll the Stripe API on every request pay for that in latency and rate limits. The webhook is the writer; everything else reads your own tables.
Four milestones carry the whole integration: a pinned, environment-driven SDK setup; a plan catalogue that maps your tiers to Stripe price identifiers without magic strings; a checkout endpoint that cannot be tricked into selling the wrong thing; and a webhook receiver that survives duplicates, retries and out-of-order delivery. Everything after that — dunning, metered usage, dashboards — hangs off those four. Line the mechanics up with your broader API pricing tier design early, because retrofitting a billing model onto a schema that only stores a boolean is_paid flag is a weekend you will not enjoy.
Prerequisites and secure SDK initialisation
Assume Python 3.11 or newer, an async-native FastAPI setup served by Uvicorn, and Postgres for state. Install stripe, fastapi, uvicorn, pydantic, python-dotenv and sqlalchemy[asyncio] with asyncpg. You need four secrets before any code runs: STRIPE_SECRET_KEY, STRIPE_WEBHOOK_SECRET, APP_BASE_URL and a database URL. In development a .env file is fine; in production read them from your platform's secret store, because a .env committed by accident is a live key in a public repository and Stripe will disable it — after somebody has already used it.
Pin the API version explicitly. Stripe evolves its API on dated versions, and an unpinned SDK upgrade can change the shape of an object you parse in a webhook handler. Pinning turns a surprise outage into a deliberate migration you schedule, exactly the same discipline you would apply when versioning and evolving public APIs of your own.
# config/stripe_client.py
import os
import stripe
from dotenv import load_dotenv
load_dotenv()
stripe.api_key = os.environ["STRIPE_SECRET_KEY"]
# Pin the dated API version so an SDK bump never reshapes a webhook payload.
stripe.api_version = os.getenv("STRIPE_API_VERSION", "2025-03-31.basil")
# Retry only idempotent-safe failures; Stripe attaches an idempotency key per retry.
stripe.max_network_retries = int(os.getenv("STRIPE_MAX_RETRIES", "3"))
APP_BASE_URL = os.environ["APP_BASE_URL"]
WEBHOOK_SECRET = os.environ["STRIPE_WEBHOOK_SECRET"]
Use os.environ[...] for anything the process cannot run without. A KeyError at import time is a loud, fast failure during deploy; os.getenv returning None is a 500 three hours later when the first customer clicks Subscribe. Reserve os.getenv with a default for genuinely optional tuning knobs, as above.
One detail trips up almost everyone: the stripe library manages its own HTTP client. Configure retries through stripe.max_network_retries and timeouts through stripe.default_http_client using Stripe's own client classes (stripe.HTTPXClient, stripe.RequestsClient). Do not assign a bare httpx.AsyncClient to it — the library expects its wrapper interface, and the failure mode is an obscure attribute error deep in a request you cannot reproduce locally. If you want async calls today, wrap the synchronous SDK call in asyncio.to_thread, which keeps the event loop free without depending on SDK internals. Recent SDK releases also expose _async method variants; to_thread is the version that works on every release you might be pinned to.
Modelling your plan catalogue
Stripe's object graph and your database schema are two views of the same product, and the mapping between them should live in exactly one file. A Product groups a sellable thing, a Price attaches a currency and interval to it, a Customer represents the buyer, and a Subscription ties a Customer to one or more Prices with a status. Your side needs a plans table your code can read without a network call, plus columns on the account row that hold the Stripe identifiers.
Keep the catalogue in a TOML file that ships with the code and reads its price identifiers from the environment. TOML gives you a reviewable diff when a price changes, tomllib is in the standard library from 3.11, and the environment indirection means the same file drives test mode and live mode without a branch. Hardcoding price_1P... strings in a route handler is how a test-mode price reaches production and starts charging nobody anything.
# plans.toml — one row per sellable tier
[starter]
price_env = "PRICE_STARTER"
monthly_usd = 9
included_requests = 10_000
[pro]
price_env = "PRICE_PRO"
monthly_usd = 29
included_requests = 100_000
[scale]
price_env = "PRICE_SCALE"
monthly_usd = 99
included_requests = 1_000_000
# billing/catalogue.py
import os
import tomllib
from dataclasses import dataclass
from functools import lru_cache
from pathlib import Path
@dataclass(frozen=True, slots=True)
class Plan:
slug: str
price_id: str
monthly_usd: int
included_requests: int
@lru_cache(maxsize=1)
def load_plans() -> dict[str, Plan]:
path = Path(os.getenv("PLANS_FILE", "plans.toml"))
raw = tomllib.loads(path.read_text(encoding="utf-8"))
plans: dict[str, Plan] = {}
for slug, cfg in raw.items():
price_id = os.environ[cfg["price_env"]] # fail loudly at boot
plans[slug] = Plan(
slug=slug,
price_id=price_id,
monthly_usd=cfg["monthly_usd"],
included_requests=cfg["included_requests"],
)
return plans
def get_plan(slug: str) -> Plan | None:
return load_plans().get(slug)
Call load_plans() once during application startup so a missing environment variable crashes the deploy rather than the first customer. The lru_cache keeps the catalogue in process memory, which removes a Stripe API round trip from your hot path entirely; the same instinct that drives caching Python API responses with Redis applies to configuration you would otherwise fetch on demand.
Generating dynamic checkout sessions
The checkout endpoint has one job: turn a plan slug the client asked for into a Stripe-hosted URL that can only sell what your catalogue says it can. Never accept a price identifier from the client. If the request body carries price_id, an attacker sends the identifier of your $0 internal test price and provisions a paid account for free. Accept a slug, look it up locally, and let the lookup fail closed.
Reuse the Stripe Customer across purchases. Creating a fresh Customer on every checkout scatters one human across five customer records, which breaks lifetime-value reporting, breaks the billing portal, and makes proration on upgrades impossible. Store stripe_customer_id on your account row the first time you create it and pass it thereafter.
# routes/payments.py
import asyncio
import stripe
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, Field
from app.auth import current_account # your API-key or JWT dependency
from billing.catalogue import get_plan
from config.stripe_client import APP_BASE_URL
router = APIRouter(prefix="/v1")
class CheckoutRequest(BaseModel):
plan: str = Field(min_length=2, max_length=32)
@router.post("/checkout")
async def create_checkout(body: CheckoutRequest, account=Depends(current_account)):
plan = get_plan(body.plan)
if plan is None:
raise HTTPException(status_code=422, detail="Unknown plan")
customer_id = account.stripe_customer_id
if customer_id is None:
customer = await asyncio.to_thread(
stripe.Customer.create,
email=account.email,
metadata={"account_id": str(account.id)},
idempotency_key=f"customer:{account.id}",
)
customer_id = customer.id
await save_customer_id(account.id, customer_id)
try:
session = await asyncio.to_thread(
stripe.checkout.Session.create,
mode="subscription",
customer=customer_id,
line_items=[{"price": plan.price_id, "quantity": 1}],
success_url=f"{APP_BASE_URL}/dashboard?session_id={{CHECKOUT_SESSION_ID}}",
cancel_url=f"{APP_BASE_URL}/pricing",
client_reference_id=str(account.id),
subscription_data={
"metadata": {"account_id": str(account.id), "plan": plan.slug},
},
metadata={"account_id": str(account.id), "plan": plan.slug},
idempotency_key=f"checkout:{account.id}:{plan.slug}",
expand=["subscription"],
)
except stripe.RateLimitError:
raise HTTPException(status_code=429, detail="Payment provider busy, retry")
except stripe.StripeError as exc:
raise HTTPException(status_code=502, detail=f"Stripe error: {exc.user_message}")
return {"checkout_url": session.url, "session_id": session.id}
Three details in that block earn their keep. The idempotency_key means a double-clicked Subscribe button returns the same session instead of creating two, and it costs nothing. Putting metadata on subscription_data as well as the session matters because renewal invoices twelve months from now carry the subscription's metadata, not the session's — without it, next year's invoice.paid event arrives with no idea which account it belongs to. And asyncio.to_thread keeps a slow Stripe call from blocking the event loop; a synchronous SDK call inside an async def route pins the whole worker for the 200-400 ms the API takes, which is the classic reason a healthy-looking service falls over at forty concurrent checkouts.
Catch stripe.RateLimitError separately and return 429 rather than 502. Stripe's live-mode limit sits around 100 read and 100 write requests per second, generous for a micro-SaaS but reachable during a launch spike, and a 429 with a Retry-After tells a well-behaved client what to do. If you drive checkout from a script or a partner integration, pair it with retrying failed HTTP requests with tenacity on the caller side.
Verifying and routing Stripe webhooks
Webhooks are the only trustworthy record of what happened. Read the raw request body — not a parsed JSON dict — because signature verification hashes the exact bytes Stripe sent, and any re-serialisation changes whitespace and breaks the HMAC. FastAPI's await request.body() gives you those bytes. The full mechanics of the Stripe-Signature header, the timestamp tolerance and the replay window are broken down in verifying Stripe webhook signatures; the summary is that construct_event rejects anything older than five minutes by default, so a badly skewed server clock produces a wall of signature failures that look like an attack.
Assume duplicates. Stripe delivers at least once, retries for up to three days with exponential backoff, and can deliver events out of order — a customer.subscription.updated can land before the checkout.session.completed that created it. Dedupe on event.id with a unique constraint, and make every handler safe to run twice. That pattern generalises well beyond Stripe; the reusable version lives in building an idempotent webhook receiver.
# routes/webhooks.py
import logging
import stripe
from fastapi import APIRouter, BackgroundTasks, Depends, Request, Response
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession
from app.db import get_session # yields an AsyncSession per request
from config.stripe_client import WEBHOOK_SECRET
router = APIRouter(prefix="/v1")
log = logging.getLogger("billing")
HANDLED = {
"checkout.session.completed",
"customer.subscription.updated",
"customer.subscription.deleted",
"invoice.payment_failed",
"invoice.paid",
}
@router.post("/webhook")
async def handle_webhook(
request: Request,
tasks: BackgroundTasks,
session: AsyncSession = Depends(get_session),
) -> Response:
payload = await request.body()
signature = request.headers.get("stripe-signature", "")
try:
event = stripe.Webhook.construct_event(payload, signature, WEBHOOK_SECRET)
except ValueError:
return Response(status_code=400, content="malformed payload")
except stripe.SignatureVerificationError:
log.warning("webhook_signature_rejected")
return Response(status_code=400, content="bad signature")
if event["type"] not in HANDLED:
return Response(status_code=200)
# Claim the event: the unique primary key makes redelivery a no-op.
claimed = await session.execute(
text(
"INSERT INTO stripe_events (id, type, received_at) "
"VALUES (:id, :type, now()) ON CONFLICT (id) DO NOTHING RETURNING id"
),
{"id": event["id"], "type": event["type"]},
)
await session.commit()
if claimed.first() is None:
log.info("webhook_duplicate", extra={"event_id": event["id"]})
return Response(status_code=200)
tasks.add_task(dispatch, event)
return Response(status_code=200)
async def dispatch(event: stripe.Event) -> None:
obj = event["data"]["object"]
match event["type"]:
case "checkout.session.completed":
await grant_entitlement(obj["metadata"]["account_id"], obj["metadata"]["plan"])
case "customer.subscription.updated":
await sync_subscription(obj["id"], obj["status"], obj["items"]["data"][0]["price"]["id"])
case "customer.subscription.deleted":
await revoke_entitlement(obj["id"])
case "invoice.payment_failed":
await start_dunning(obj["customer"], obj["attempt_count"])
case "invoice.paid":
await clear_dunning(obj["customer"])
case unexpected:
log.info("webhook_ignored", extra={"type": unexpected})
The match statement is not decoration here. Stripe emits well over two hundred event types and your allow-list will grow; a match with a case _ fallback that logs the type gives you a readable audit of what you are choosing not to handle, and the structural patterns compose neatly when you later need to branch on obj["status"] inside a case.
Note the shape of the response path. The endpoint commits the dedupe row, hands the real work to a background task, and returns 200 in single-digit milliseconds. Stripe's delivery timeout is short, and an endpoint that takes eight seconds to write to Postgres and send a welcome email will be marked as failing, retried, and eventually disabled. For work heavier than a couple of queries — provisioning infrastructure, generating an API key bundle, calling a CRM — push the job onto a real queue instead, as covered in running background jobs with Celery. FastAPI's BackgroundTasks runs in-process and dies with the worker; a queue survives a restart.
Return 200 for events you do not handle, and 400 only for genuinely malformed or unsigned input. Returning 500 for an unknown event type teaches Stripe to retry something you will never process, and after enough failures Stripe disables the endpoint for the whole account — including the events you do care about.
Tracking the subscription lifecycle
A subscription is a state machine, and your entitlement check should read one column rather than reasoning about invoices. Stripe reports trialing, active, past_due, canceled, unpaid and incomplete; you mirror the ones that matter into an accounts.status column and gate API access on it. Treat trialing and active as serving states, past_due as a serving-with-warning state for a defined grace window, and everything else as revoked.
Upgrades and downgrades run through Subscription.modify, and the proration setting is a pricing decision, not a technical one. create_prorations charges the difference immediately and is right for upgrades — the customer wanted more capacity now and expects to pay for it. For downgrades, none combined with an end-of-period schedule avoids issuing credits you will later have to explain. Never delete and recreate a subscription to change a plan: you lose the billing anchor, reset the invoice history, and hand the customer a full-price charge on a day they did not expect one.
# billing/lifecycle.py
import asyncio
import stripe
from billing.catalogue import get_plan
async def change_plan(subscription_id: str, new_slug: str, *, upgrade: bool) -> stripe.Subscription:
plan = get_plan(new_slug)
if plan is None:
raise ValueError(f"unknown plan: {new_slug}")
sub = await asyncio.to_thread(stripe.Subscription.retrieve, subscription_id)
item_id = sub["items"]["data"][0]["id"]
return await asyncio.to_thread(
stripe.Subscription.modify,
subscription_id,
items=[{"id": item_id, "price": plan.price_id}],
proration_behavior="create_prorations" if upgrade else "none",
metadata={"plan": plan.slug},
idempotency_key=f"plan-change:{subscription_id}:{plan.slug}",
)
Do not write the new tier to your database here. Let the customer.subscription.updated webhook do it. If you write locally and the Stripe call later fails a retry, your database claims the customer is on Scale while Stripe still bills them for Starter — a discrepancy nobody notices until a support ticket arrives months later. One writer, always. Verifying that this holds across a full year of renewals is exactly what Stripe test clocks exist for.
For usage-priced tiers, the same subscription carries metered items and you report quantities on a schedule instead of at checkout. The reporting cadence, aggregation mode and rounding rules are their own topic, handled in Stripe metered billing configuration, and the counters that feed it come from tracking API usage and analytics.
Configuration reference
Every value below belongs in the environment, and the production column is the setting I would ship rather than the SDK default.
| Variable | Default | Production setting |
|---|---|---|
STRIPE_SECRET_KEY | none | Restricted key, write scope on billing only |
STRIPE_WEBHOOK_SECRET | none | Per-endpoint secret, rotated yearly |
STRIPE_API_VERSION | SDK default | Pinned dated version, bumped deliberately |
STRIPE_MAX_RETRIES | 0 | 3 |
APP_BASE_URL | none | Public HTTPS origin, no trailing slash |
PLANS_FILE | plans.toml | Baked into the image, read-only |
WEBHOOK_GRACE_DAYS | none | 5 for monthly, 10 for annual plans |
Use a restricted key rather than the account secret key. A restricted key scoped to write Checkout Sessions and Subscriptions cannot issue refunds or read your payout schedule, which caps the blast radius when a container image leaks. Rotate the webhook secret by adding a second endpoint in the dashboard, running both for a day, then deleting the old one — the same zero-downtime discipline as rotating API keys without downtime on your own service.
Verifying the integration end to end
Do not trust an integration you have only seen work in a browser. Run the Stripe CLI against your local server and confirm the signature path, the dedupe path and the entitlement write in one pass.
# terminal 1 — forward live test-mode events to your local endpoint
export STRIPE_WEBHOOK_SECRET=$(stripe listen --print-secret)
stripe listen --forward-to localhost:8000/v1/webhook
# terminal 2 — drive the flow
curl -sS -X POST localhost:8000/v1/checkout \
-H "Authorization: Bearer $DEV_API_TOKEN" \
-H 'content-type: application/json' \
-d '{"plan":"pro"}' | python3 -m json.tool
# replay the same event twice: the second must log webhook_duplicate
stripe trigger checkout.session.completed
stripe trigger checkout.session.completed
The card 4242 4242 4242 4242 succeeds, 4000 0000 0000 9995 fails with insufficient_funds, and 4000 0025 0000 3155 forces a 3D Secure challenge. Run all three. The 3DS card is the one that catches teams who assumed the session completes synchronously — the customer authenticates, the browser returns, and the subscription sits in incomplete until the bank confirms, which can take several seconds.
In the test suite, do not call Stripe. Stub the HTTP layer so your tests stay fast and deterministic, using the approach in mocking external APIs with respx, and keep a small set of signed-payload fixtures for the webhook route. Generate those fixtures with stripe.Webhook.construct_event's counterpart — sign a payload with a known test secret using hmac — so the signature branch is covered rather than skipped.
Cost, margin and performance at scale
Stripe charges 2.9% plus $0.30 per successful card charge in the US, and that fixed 30 cents is the number that decides whether a cheap tier is viable. At $5 a month you surrender 8.9% of revenue to fees; at $99 you surrender 3.2%. The compute cost of the integration itself is noise by comparison — a webhook handler that commits one row and returns runs in about 4 ms of CPU, so ten thousand billing events a month cost well under a cent of compute on any platform you would actually deploy to.
The commercial conclusion is blunt: bill anything under about $15 annually rather than monthly. A $9 monthly plan gives up $6.73 a year in fees; the same $108 taken once a year gives up $3.43. That is roughly 3% of revenue recovered for a config change, and annual billing also removes eleven chances a year for the card to decline. Model the whole picture — fees, compute, and the support cost of a failed charge — with the method in calculating cost per API request.
On the performance side, size the webhook endpoint for burst, not average. A monthly renewal batch fires most of your invoice.paid events inside a few minutes on the same day of the month, so a service that handles two events an hour on average may see two hundred in sixty seconds. Because the handler is I/O-bound and short, two Uvicorn workers per vCPU absorb that comfortably; the real constraint is your database connection pool, and a pool of five behind a burst of two hundred concurrent inserts is the classic cause of connection pool exhaustion. Keep the dedupe insert in its own short transaction, as above, and it holds a connection for about a millisecond.
Gotchas and failure modes
- Granting access on the success redirect. The
success_urlruns in the customer's browser and can be replayed by anyone who saves the link. Treat it as a "thanks, we're setting things up" page and let the webhook do the granting. - Parsing the body before verifying the signature. Any framework middleware that re-encodes JSON invalidates the HMAC. Read raw bytes, verify, then parse.
- Metadata only on the checkout session. Renewal invoices carry subscription metadata. Without
subscription_data.metadatayou cannot map next month's event to an account without an extra API call. - Synchronous SDK calls inside async routes. A blocking
Session.createinsideasync deffreezes the event loop for the whole worker. Useasyncio.to_threador a threadpool route. - Ignoring
customer.subscription.deleted. A cancelled customer whose entitlement row never flips keeps hitting your paid endpoints for free, and this is the single most common source of silent margin leak in early micro-SaaS. - One webhook endpoint for test and live mode. The secrets differ, so half your events fail verification and you conclude the integration is broken.
FAQ
What does Stripe actually cost me at $10,000 in monthly recurring revenue? On 345 customers paying $29, fees run about $404 a month — $290 from the 2.9% and $104 from the fixed $0.30 per charge. Compute for the billing path is under a dollar. Moving those customers to annual billing at $290 cuts total fees to roughly $187 a year per hundred customers, so the same revenue keeps about $200 more each month.
Should I use Stripe Checkout or Stripe Elements for an API-first product? Checkout, almost always. It keeps you in the lightest PCI scope (SAQ A), handles 3D Secure, tax collection, and wallet payments without extra code, and ships in an afternoon. Elements only earns its complexity when the payment step must live inside your own UI and you have the frontend capacity to maintain it.
How do I migrate off Stripe later without breaking billing? Store your own plan slugs and entitlement status as the source of truth, keep Stripe identifiers in dedicated nullable columns, and never let a Stripe object identifier become a foreign key in your domain tables. Stripe will export card data to another PCI-compliant provider on request, so the migration cost is your integration code, not your customers' cards — usually a week if the coupling is thin.
How often should I rotate the webhook signing secret? Yearly, or immediately after anyone with access leaves. Add a second endpoint with a new secret, deploy code that accepts either, delete the old endpoint after a day, then drop the fallback. Downtime is zero because Stripe signs per endpoint.
Do I need to handle every one of Stripe's event types?
No, and trying to is a trap. Five events cover a subscription business: checkout.session.completed, customer.subscription.updated, customer.subscription.deleted, invoice.payment_failed and invoice.paid. Subscribe to those in the dashboard so you are not paying to receive noise, and return 200 for anything else that arrives.
Related
Same topic area:
- Handling failed payments and dunning — the recovery sequence after
invoice.payment_failedfires. - Testing Stripe integrations with test clocks — compress a year of renewals into a two-minute test run.
- Designing API pricing tiers — decide what the plan catalogue above should contain.
- Best platforms to host Python APIs for free — where to run the webhook endpoint before revenue justifies paid infrastructure.
Adjacent areas:
- Verifying Stripe webhook signatures — the HMAC and replay-window details behind
construct_event. - Syncing Stripe customers to HubSpot with Python — push the same lifecycle events into your CRM.