""" Pro hybrid billing (committed base + metered overage) — Stripe Billing Meters sample. Grounded in official Stripe docs (see refs/stripe-usage-billing.md for source URLs, cited 2026-07-07): - POST /v1/billing/meters stripe.billing.Meter.create (S2) - POST /v1/billing/meter_events stripe.billing.MeterEvent.create (S3, dedup via `identifier`) - POST /v1/prices stripe.Price.create (S5, metered + tiered overage) - POST /v1/subscriptions stripe.Subscription.create (S6, licensed base + metered item) This is a reference implementation of the backend metering + billing slice. It is runnable against a Stripe *test* account (STRIPE_API_KEY=sk_test_...). No network calls happen at import time. pip install stripe export STRIPE_API_KEY=sk_test_xxx python metering_sample.py --demo # provisions meter/prices in the test account Design intent: * The committed base fee is a LICENSED price (fixed, predictable ARR). * Overage is a METERED price whose graduated tier-1 (the included allotment) costs 0, so only usage above the allotment is billed. Stripe does the aggregation + tiering; the app only emits idempotent meter events. """ from __future__ import annotations import argparse import hashlib import os import uuid from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Optional import stripe stripe.api_key = os.environ.get("STRIPE_API_KEY", "") # --------------------------------------------------------------------------- # Config — the value-metric + pricing parameters (to be finalized by CPO/CFO, # per exec packet user-decision-needed). Amounts are in the smallest currency unit. # --------------------------------------------------------------------------- CURRENCY = "usd" EVENT_NAME = "pro_api_call" # the value-metric event; keep <=100 chars, stable forever AGGREGATION_FORMULA = "sum" # sum | count | last (sum of `value` per period) INCLUDED_ALLOTMENT = 10_000 # units covered by the committed base fee (tier-1 = 0) BASE_FEE_AMOUNT = 2_000 # $20.00/mo committed base (licensed) OVERAGE_UNIT_AMOUNT = 2 # $0.02 per unit above the allotment (unit-margin floor guarded) # --------------------------------------------------------------------------- # 1. Provision the billing primitives (run once per environment) # --------------------------------------------------------------------------- @dataclass class ProPlan: meter_id: str product_id: str base_price_id: str metered_price_id: str def create_meter() -> stripe.billing.Meter: """POST /v1/billing/meters — defines how usage events aggregate over a period. (S2)""" return stripe.billing.Meter.create( display_name="Pro API Calls", event_name=EVENT_NAME, default_aggregation={"formula": AGGREGATION_FORMULA}, value_settings={"event_payload_key": "value"}, customer_mapping={"type": "by_id", "event_payload_key": "stripe_customer_id"}, ) def create_base_price(product_id: str) -> stripe.Price: """POST /v1/prices — licensed committed base fee (predictable ARR). (S5)""" return stripe.Price.create( currency=CURRENCY, product=product_id, unit_amount=BASE_FEE_AMOUNT, recurring={"interval": "month"}, # default usage_type = licensed ) def create_metered_overage_price(product_id: str, meter_id: str) -> stripe.Price: """POST /v1/prices — metered price with graduated tiers: allotment free, overage billed. (S5)""" return stripe.Price.create( currency=CURRENCY, product=product_id, recurring={"interval": "month", "usage_type": "metered", "meter": meter_id}, billing_scheme="tiered", tiers_mode="graduated", tiers=[ {"up_to": INCLUDED_ALLOTMENT, "unit_amount": 0}, # included in base fee {"up_to": "inf", "unit_amount": OVERAGE_UNIT_AMOUNT}, # overage ], ) def provision_pro_plan() -> ProPlan: meter = create_meter() product = stripe.Product.create(name="Pro") base_price = create_base_price(product.id) metered_price = create_metered_overage_price(product.id, meter.id) return ProPlan( meter_id=meter.id, product_id=product.id, base_price_id=base_price.id, metered_price_id=metered_price.id, ) # --------------------------------------------------------------------------- # 2. Subscribe a customer to the hybrid plan # --------------------------------------------------------------------------- def subscribe_customer(customer_id: str, plan: ProPlan) -> stripe.Subscription: """ POST /v1/subscriptions — one subscription, two items. (S6) NOTE: the metered item must NOT carry a `quantity`; usage is reported via meter events. The Idempotency-Key header makes the create safe to retry. """ return stripe.Subscription.create( customer=customer_id, items=[ {"price": plan.base_price_id, "quantity": 1}, # licensed committed fee {"price": plan.metered_price_id}, # metered overage — no quantity ], idempotency_key=f"sub-create:{customer_id}:pro-v1", ) # --------------------------------------------------------------------------- # 3. Record usage — the hot path. Idempotent by construction. # --------------------------------------------------------------------------- def _dedup_identifier(customer_id: str, unit_key: str) -> str: """ Deterministic identifier so at-least-once delivery / retries never double-count. Stripe enforces uniqueness within a rolling 24h window (S4), so `unit_key` must be unique per real-world usage unit within that window (e.g. request id, job id). """ raw = f"{EVENT_NAME}:{customer_id}:{unit_key}" return hashlib.sha256(raw.encode()).hexdigest()[:64] def record_usage( customer_id: str, value: int, unit_key: str, when: Optional[datetime] = None, ) -> stripe.billing.MeterEvent: """ POST /v1/billing/meter_events — report one usage event. (S3) * `identifier` = deterministic dedup key (idempotency within 24h). * `timestamp` = event time; Stripe accepts within past 35 days / +5 min (S4). Defaults to now. """ payload = {"stripe_customer_id": customer_id, "value": str(value)} kwargs = { "event_name": EVENT_NAME, "payload": payload, "identifier": _dedup_identifier(customer_id, unit_key), } if when is not None: kwargs["timestamp"] = int(when.replace(tzinfo=timezone.utc).timestamp()) return stripe.billing.MeterEvent.create(**kwargs) # --------------------------------------------------------------------------- # 4. Overage estimate (UX-only). Stripe remains the ledger of record. # --------------------------------------------------------------------------- @dataclass class UsageEstimate: aggregated_usage: int included_allotment: int = INCLUDED_ALLOTMENT overage_unit_amount: int = OVERAGE_UNIT_AMOUNT base_fee_amount: int = BASE_FEE_AMOUNT @property def overage_units(self) -> int: return max(0, self.aggregated_usage - self.included_allotment) @property def estimated_overage_charge(self) -> int: return self.overage_units * self.overage_unit_amount @property def estimated_invoice_total(self) -> int: # Single invoice at renewal: committed base + metered overage (S6). return self.base_fee_amount + self.estimated_overage_charge def estimate_overage(aggregated_usage: int) -> UsageEstimate: """ Client-facing estimate for 60/80/100% nudges. This mirrors Stripe's tiering math for UX only; the authoritative amount is computed by Stripe at invoice finalization. """ return UsageEstimate(aggregated_usage=aggregated_usage) # --------------------------------------------------------------------------- # Simple consuming-team interface (product/frontend call these two, nothing else) # --------------------------------------------------------------------------- @dataclass class MeteringClient: """Thin facade the rest of the product uses. Hides Stripe API surface + idempotency.""" plan: ProPlan _seen: set = field(default_factory=set) # optional local guard; Stripe is source of truth def report(self, customer_id: str, value: int, unit_key: str) -> None: record_usage(customer_id=customer_id, value=value, unit_key=unit_key) def estimate(self, customer_id: str, aggregated_usage: int) -> UsageEstimate: return estimate_overage(aggregated_usage) # --------------------------------------------------------------------------- # Demo # --------------------------------------------------------------------------- def _demo() -> None: if not stripe.api_key: raise SystemExit("Set STRIPE_API_KEY=sk_test_... to run the demo.") plan = provision_pro_plan() print("Provisioned:", plan) customer = stripe.Customer.create(name="Demo Co") sub = subscribe_customer(customer.id, plan) print("Subscription:", sub.id, "status:", sub.status) # Emit a couple of idempotent usage events (retry-safe by unit_key). req_id = str(uuid.uuid4()) record_usage(customer.id, value=25, unit_key=req_id) record_usage(customer.id, value=25, unit_key=req_id) # duplicate -> deduped by identifier est = estimate_overage(aggregated_usage=12_500) print(f"Est. overage units={est.overage_units} charge={est.estimated_overage_charge} " f"invoice_total={est.estimated_invoice_total}") if __name__ == "__main__": parser = argparse.ArgumentParser(description="Stripe metered-overage sample") parser.add_argument("--demo", action="store_true", help="provision + run against test account") args = parser.parse_args() if args.demo: _demo() else: print(__doc__)