Python API Subscription Billing Tutorial: Stripe Integration and Usage Tracking

Implementing subscription billing for a commercial Python API is not a payment-form problem — it is an access-control problem with money attached. You need cryptographic webhook verification, idempotent event processing, and real-time quota enforcement working together, because any gap between "customer paid" and "customer can call the endpoint" is either revenue you never collect or compute you give away for free. This walkthrough sits under the Building API Marketplaces guide and delivers a production architecture for tiered subscriptions and metered usage using FastAPI, Stripe, and Redis. It is the billing spine behind everything in the wider Building & Monetizing API-Driven Micro-SaaS section.

The four building blocks this tutorial wires together:

  • Checkout sessions mapped to internal pricing tiers via metadata
  • Cryptographically verified webhook signatures
  • Idempotent event handling that eliminates duplicate provisioning
  • Redis-backed atomic usage counters with hard quota enforcement

The subscription itself is a small state machine. Model it explicitly before you write a line of billing code — every webhook you handle is just a transition in this diagram, and every bug is a transition you forgot to handle.

Subscription lifecycle states A subscriber moves from signup to active, renews each cycle, and on a failed payment is suspended or cancelled. Signup Active Renew or fail Cancel / suspend successful renewal loops back to active

1. Architecture and Prerequisites

Before writing billing logic, establish a resilient stack. Two moving parts do the heavy lifting: FastAPI for the request path and Redis for the hot state that a relational database is too slow to serve on every call.

Bash
pip install fastapi uvicorn stripe redis pydantic python-dotenv

Configure environment variables and never hardcode secrets or price IDs. Price IDs in particular change whenever you re-tier your pricing, and a hardcoded one silently routes new customers to the wrong plan.

Env
STRIPE_SECRET_KEY=sk_test_...
STRIPE_WEBHOOK_SECRET=whsec_...
REDIS_URL=redis://localhost:6379/0
PRICE_STARTER=price_1Nq...
PRICE_PRO=price_1Nr...

Redis serves two roles that decide the whole design. First, atomic request counting via INCR, which returns the post-increment value in a single round trip so two concurrent requests can never both read "999" and both pass a 1000-call quota. Second, distributed idempotency tracking for webhook events. Reach for Redis here rather than Postgres because the counter is written on every single request — pushing that write into your primary database couples billing throughput to your slowest table and burns connections you need for real work. Postgres remains the system of record for who owns which subscription; Redis is the fast, disposable layer in front of it. If Redis is new to your stack, the setup and eviction tuning are covered in caching Python API responses with Redis.

Configure the instance with maxmemory-policy allkeys-lru so stale event IDs and expired counters evict cleanly instead of pushing the box into an out-of-memory stall. Size maxmemory to leave headroom: at a few million distinct keys with 24-hour TTLs you are still well inside a 512 MB plan.

2. Implementing Tiered Subscription Checkout

Create a dedicated FastAPI route that generates Stripe Checkout sessions and attaches tenant metadata, so a successful payment maps back to an internal user record. The tier-to-price decision belongs in code you control, not in the client request — the caller names a tier, never a raw price ID. The pricing model behind these tiers is worked through in how to charge for API access using Stripe.

Python
import os
import stripe
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel

router = APIRouter()
stripe.api_key = os.getenv("STRIPE_SECRET_KEY")

class SubscribeRequest(BaseModel):
    user_id: str
    tier: str  # 'starter' or 'pro'

@router.post("/billing/subscribe")
async def create_checkout(req: SubscribeRequest):
    price_map = {
        "starter": os.getenv("PRICE_STARTER"),
        "pro": os.getenv("PRICE_PRO"),
    }

    if req.tier not in price_map:
        raise HTTPException(status_code=400, detail="Invalid pricing tier")

    try:
        session = stripe.checkout.Session.create(
            payment_method_types=["card"],
            line_items=[{"price": price_map[req.tier], "quantity": 1}],
            mode="subscription",
            success_url=f"{os.getenv('APP_BASE_URL')}/billing/success?session_id={{CHECKOUT_SESSION_ID}}",
            cancel_url=f"{os.getenv('APP_BASE_URL')}/billing/cancel",
            client_reference_id=req.user_id,
            metadata={"user_id": req.user_id},
            subscription_data={"metadata": {"user_id": req.user_id}},
        )
        return {"checkout_url": session.url}
    except stripe.StripeError as e:
        raise HTTPException(status_code=502, detail=f"Stripe API failure: {str(e)}")

Why this works: Pydantic validates the request, the tier maps to a price ID from the environment, and user_id rides along in two places. Setting subscription_data.metadata is the detail most tutorials miss — metadata on the Checkout Session does not propagate to the resulting subscription object, so later events like invoice.payment_failed and customer.subscription.deleted would arrive with no way to identify the customer. Duplicating it into subscription_data guarantees every downstream event carries user_id.

3. Metered Usage Tracking and Quota Enforcement

Hard limits stop abuse; soft limits degrade gracefully. Use middleware to intercept each request, atomically increment the counter, and reject over-quota traffic before the route runs any expensive work. The data path is a short pipeline with one asynchronous branch that keeps Stripe in sync without sitting on the request.

Metered usage request path and background sync Each request increments a Redis counter and passes a quota gate; a separate hourly worker reads counters and reports usage to Stripe. request path (synchronous) API request Auth + tenant resolve user_id Redis INCR usage:{user} Quota gate Route over limit → 402 background path (hourly, off the request) Hourly sync worker Stripe UsageRecord
Python
import os
import redis.asyncio as redis
from fastapi import Request
from fastapi.responses import JSONResponse

redis_client = redis.Redis.from_url(os.getenv("REDIS_URL", "redis://localhost:6379/0"), decode_responses=True)

async def get_user_daily_limit(user_id: str) -> int:
    # Replace with actual DB/cache lookup
    return 1000  # Example: starter tier

async def log_overage_event(user_id: str, overage: int) -> None:
    # Push to analytics queue for Stripe UsageRecord sync
    pass

async def usage_middleware(request: Request, call_next):
    user_id = getattr(request.state, "user_id", None)
    if not user_id:
        return await call_next(request)

    limit = await get_user_daily_limit(user_id)

    try:
        current = await redis_client.incr(f"usage:{user_id}")
        if current == 1:
            await redis_client.expire(f"usage:{user_id}", 86400)  # 24h TTL

        if current > limit:
            await log_overage_event(user_id, current - limit)
            return JSONResponse(
                status_code=402,
                content={"error": "Quota exceeded. Upgrade your plan or wait for reset."},
            )
    except redis.ConnectionError:
        # Fail closed to protect infrastructure
        return JSONResponse(status_code=503, content={"error": "Usage tracking unavailable"})

    return await call_next(request)

There is a subtle race worth naming: the INCR and the EXPIRE are two commands, so a crash between them leaves a counter with no TTL that never resets. For strict correctness, move both into a Lua script or a SET key 0 EX 86400 NX seed followed by INCR, making the window atomic. For most APIs the two-command form is fine because the next day's first request re-seeds the key anyway — know the trade-off and pick deliberately.

Fail closed, not open. When Redis is unreachable the middleware returns 503 rather than waving traffic through. An API that fails open during a Redis outage is one bad afternoon away from serving your entire paid workload for free. If you would rather degrade to a coarser limit than reject outright, that belongs in the same free-tier abuse prevention logic.

Never call Stripe on the request path. The log_overage_event hook writes to a queue; a separate hourly worker reads counters and reports them to Stripe's UsageRecord API. Calling Stripe per request adds 100–300 ms of latency and burns straight through Stripe's rate limits. The metered-price setup and reporting cadence are detailed in Stripe metered billing configuration, and if you also want a durable audit trail, mirror each counter flush into Postgres as described in logging API usage events to Postgres.

4. Webhook Handling and Idempotent Event Processing

Stripe retries any webhook it does not see acknowledged with a 2xx, with exponential backoff over up to three days. Without idempotency you will double-provision access or fire duplicate billing side effects. Verify the stripe-signature header, then dedupe on event["id"] with an atomic Redis SADD. The full signing-secret discipline is covered in verifying Stripe webhook signatures; the general receiver pattern in building an idempotent webhook receiver.

Idempotent webhook processing with retry deduplication Stripe delivers an event; the endpoint records its id in a Redis set and provisions once. A retry of the same event is recognised as a duplicate and acknowledged without re-provisioning. Stripe API endpoint Redis set delivery 1: POST event SADD event.id → 1 (new) provision access 200 OK retry: same event id SADD event.id → 0 (dup) 200 already_processed (no re-provision)
Python
import os
import stripe
from fastapi import Request, HTTPException

@router.post("/billing/webhook")
async def handle_webhook(request: Request):
    payload = await request.body()
    sig_header = request.headers.get("stripe-signature")

    try:
        event = stripe.Webhook.construct_event(
            payload, sig_header, os.getenv("STRIPE_WEBHOOK_SECRET")
        )
    except ValueError:
        raise HTTPException(status_code=400, detail="Invalid JSON payload")
    except stripe.SignatureVerificationError:
        raise HTTPException(status_code=400, detail="Invalid webhook signature")

    # Idempotency guard: sadd returns 0 if the member already exists
    if await redis_client.sadd("processed_webhooks", event["id"]) == 0:
        return {"status": "already_processed"}

    try:
        match event["type"]:
            case "customer.subscription.created":
                obj = event["data"]["object"]
                await activate_subscription(obj["metadata"]["user_id"], obj["id"])
            case "invoice.payment_failed":
                user_id = event["data"]["object"]["metadata"].get("user_id")
                if user_id:
                    await downgrade_or_suspend(user_id)
            case "customer.subscription.deleted":
                user_id = event["data"]["object"]["metadata"].get("user_id")
                if user_id:
                    await revoke_api_access(user_id)
    except Exception:
        # Roll back the idempotency marker so Stripe's retry can re-run cleanly.
        await redis_client.srem("processed_webhooks", event["id"])
        raise HTTPException(status_code=500, detail="Provisioning failed")

    return {"status": "success"}

Security critical: return 2xx only after successful processing. Returning 200 on failure tells Stripe to stop retrying and freezes your system in an inconsistent state. Note the srem on failure — because the sadd marks the event processed before the work runs, a mid-provision crash would otherwise permanently suppress the retry. Removing the marker on the exception path restores at-least-once delivery. Give the processed_webhooks set a bounded TTL or periodic trim; Stripe never retries beyond three days, so anything older is dead weight. Before you ship, exercise the failure branches against Stripe's simulated clock — see testing Stripe integrations with test clocks.

5. Cost and Performance at Scale

The architecture earns its keep on the numbers. Redis INCR is a sub-millisecond operation; a pipelined batch is faster still per call. Reporting usage to Stripe on the request path, by contrast, drags a real HTTPS round trip and Stripe's own processing into every response. The gap is roughly two orders of magnitude — which is exactly why the counter lives in Redis and the Stripe sync lives in an hourly worker.

Added latency per request by usage-tracking approach Pipelined Redis increment adds about 0.3 milliseconds, a single Redis increment about 1.2, and a per-request Stripe API call about 180 milliseconds. Added p95 latency per request (ms) — note the ~150x gap Redis INCR (pipelined) 0.3 Redis INCR (single) 1.2 Stripe API per request 180 Small bars are widened to stay visible; labels carry the true values.

Put commercial numbers on it. At 1M requests a month the usage layer costs almost nothing to run: a single small Redis instance (roughly 5–15 USD/month on a managed plan) plus negligible extra compute, since INCR never touches your worker's CPU in any meaningful way. Stripe's cut is your real cost of goods — percentage-plus-fixed per successful charge, not per API call — so batching usage keeps your Stripe API-call volume flat regardless of traffic. Model the full unit economics with calculating cost per API request, and surface the counters you are already collecting to customers via tracking API usage and analytics. The margin lesson is blunt: keep billing state in Redis, keep Stripe calls batched, and the billing subsystem stays a rounding error against your compute bill.

6. Deployment and Production Troubleshooting

Deploy the FastAPI app anywhere that gives you a stable HTTPS URL for the webhook — the trade-offs between hosts are compared in deploying APIs to Render or Vercel. Register the endpoint in the Stripe Dashboard pointing at https://yourdomain.com/billing/webhook, and roll new versions with zero-downtime deploys so an in-flight webhook is never dropped mid-restart.

Local testing workflow:

  1. Install the Stripe CLI: brew install stripe/stripe-cli/stripe
  2. Forward events locally: stripe listen --forward-to localhost:8000/billing/webhook
  3. Trigger test events: stripe trigger checkout.session.completed

Common runtime failures and fixes:

  • Signature mismatch: STRIPE_WEBHOOK_SECRET must be the exact endpoint secret from the Dashboard (or the one the CLI prints on listen), never the account secret key. Each endpoint has its own secret.
  • Timeout errors: Stripe expects a response within its delivery window. Offload heavy provisioning to a background queue and acknowledge fast.
  • Duplicate provisioning: the sadd idempotency check must run before any state mutation, and the srem rollback must run on failure.

Common Mistakes

  • Skipping webhook signature verification: leaves the API open to forged payment events and free tier upgrades from anyone who finds the URL.
  • Ignoring invoice.payment_failed: hands non-paying customers continued access. Suspend or downgrade promptly and pair it with a retry schedule — see handling failed payments and dunning.
  • Synchronous DB or Stripe calls in middleware: blocks the event loop. Keep the request path on redis.asyncio and push everything slow to a worker.
  • Hardcoding price IDs: silently misroutes customers when you re-tier. Load them from the environment.
  • Metadata only on the Checkout Session: downstream subscription events arrive with no user_id. Set subscription_data.metadata too.

FAQ

How much does subscription billing cost to run at 1M requests a month? The usage layer is a rounding error: one small managed Redis instance (about 5–15 USD/month) plus negligible worker compute, because INCR and the hourly Stripe sync barely register on CPU. Your real cost is Stripe's percentage-plus-fixed fee per successful charge, which scales with revenue, not with API traffic — so batching usage keeps that cost flat as calls grow.

What is the most reliable way to handle Stripe webhook retries in Python? Verify the signature, then dedupe on event["id"] with an atomic Redis SADD before touching any state, and roll the marker back with SREM if provisioning throws. Return 2xx only after success. That combination gives you exactly-once side effects on top of Stripe's at-least-once delivery.

Can I track metered usage in Stripe without Redis? You can, but do not in production. A Stripe API call per request adds 100–300 ms of latency and burns through rate limits within a modest traffic spike. Count atomically in Redis on the request path and batch-report to Stripe UsageRecords hourly from a worker — roughly 150x less latency per call.

How do I stop customers from using the API during the webhook processing delay? Provision on the fast signal — grant access on checkout.session.completed rather than waiting for the first invoice.paid — but scope new subscribers to a provisional tier with tight rate limits until payment confirms. That closes the free-access window without blocking legitimate paying users at signup.

What happens to billing if I need to rotate my Stripe or Redis credentials? Stripe keys roll without downtime because verification reads STRIPE_WEBHOOK_SECRET from the environment at request time — deploy the new value and restart workers. Redis credential rotation is riskier: drain in-flight requests first, because a mid-rotation ConnectionError triggers the fail-closed 503. Rotate during a low-traffic window and confirm the new URL resolves before cutting over.

Same section:

Adjacent areas: