"""Stripe: the only place money is handled, and the thinnest wrapper that can be.
This app never sees a card. Checkout and the customer portal are Stripe-hosted
pages (the browser is redirected to them and back), and the subscription's true
state arrives as signed webhook events - so this module's job is just to start
those hosted flows and to fold a verified event back into the local Subscription
row. Everything downstream reads the row, never Stripe.
Billing is *optional*. When STRIPE_SECRET_KEY is unset, is_enabled() is False,
the endpoints report billing unavailable, and every org stays free - so the app
(sign-up, onboarding, the whole shell) runs locally and in CI with no Stripe
account, the same "unset config = feature off" stance mail.py takes for SMTP.
Nothing here raises an HTTP error or touches db_session directly beyond the
session it is handed; it raises Invalid for a misconfiguration a façade turns
into a message.
"""
import logging
import time
from datetime import datetime, timezone
import settings
from db.db_declarations import Organization, Subscription
from . import plans
from .context import Invalid
logger = logging.getLogger(__name__)
# `stripe` is imported lazily inside the functions that actually call its API, not
# at module load. So the app starts and the whole test suite runs with the
# package absent (billing simply off), and only a live Stripe call - which only
# happens when billing is configured - needs it installed. is_enabled() and
# sync_from_event() never touch it.
def is_enabled() -> bool:
"""Billing is live only when a secret key is configured. Everything that
would call Stripe checks this first and degrades to 'stay free' when False."""
return bool(settings.Config.STRIPE_SECRET_KEY)
def _client():
"""Set the api key from config on each use rather than at import, so a key
rotated in the environment is picked up without a restart and tests that set
it late still work. Returns the stripe module, configured."""
if not is_enabled():
raise Invalid("Billing is not configured on this deployment.")
import stripe
stripe.api_key = settings.Config.STRIPE_SECRET_KEY
return stripe
# --- tier <-> Stripe price mapping ------------------------------------------
def _price_map() -> dict:
"""The paid tiers' Stripe price ids, from config. Keyed by tier; a tier with
no configured price is omitted, so buying it fails cleanly rather than
starting a checkout for a price that does not exist."""
cfg = settings.Config
out = {}
if cfg.STRIPE_PRICE_SILVER:
out["silver"] = cfg.STRIPE_PRICE_SILVER
if cfg.STRIPE_PRICE_GOLD:
out["gold"] = cfg.STRIPE_PRICE_GOLD
return out
def price_for_tier(tier: str) -> str:
price = _price_map().get(tier)
if not price:
raise Invalid(f"No Stripe price is configured for the {tier!r} plan.")
return price
def tier_for_price(price_id: str) -> str | None:
"""Reverse the map: which tier a Stripe price belongs to, or None if it is
one we do not recognise (a price created outside this app)."""
for tier, pid in _price_map().items():
if pid == price_id:
return tier
return None
# --- what a tier really costs -----------------------------------------------
#
# plans.py declares a price so the app can render one with no network. Stripe
# holds the amount that is actually charged, and a card that disagrees with the
# checkout it leads to is worse than a card with no price at all - so when Stripe
# is configured, its number wins.
#
# Cached for the whole process: a price changes about never, and no page load
# should wait on two Stripe round-trips to draw a plan card. A failed read is
# cached the same way, so a wrong price id costs one timeout per TTL rather than
# one per visitor.
_PRICE_TTL_S = 600.0
_price_cache: dict = {"at": None, "prices": {}}
def clear_price_cache() -> None:
"""Forget the cached amounts, so the next read goes back to Stripe. For tests,
and for the deploy that changes a price id without restarting the process."""
_price_cache["at"] = None
_price_cache["prices"] = {}
def live_prices() -> dict:
"""The configured Stripe Prices' real amounts, keyed by tier:
{"silver": {"amount": 900, "currency": "CHF", "interval": "month"}}
Empty when billing is off, when a tier has no price id, or when Stripe cannot
be reached. Every caller reads "no live price" as "show the declared one", so
a Stripe outage costs a possibly stale number and never a broken page - which
is why nothing here raises.
"""
if not is_enabled():
return {}
now = time.monotonic()
cached_at = _price_cache["at"]
if cached_at is not None and now - cached_at < _PRICE_TTL_S:
return _price_cache["prices"]
out = {}
for tier, price_id in _price_map().items():
try:
price = _client().Price.retrieve(price_id)
except Exception:
logger.warning("Could not read the Stripe price for the %s plan (%s); "
"falling back to the declared amount.", tier, price_id)
continue
amount = price.get("unit_amount")
if amount is None:
# A tiered or metered price has no single number to put on a card.
logger.warning("The Stripe price for the %s plan (%s) has no flat "
"unit_amount; falling back to the declared amount.",
tier, price_id)
continue
out[tier] = {
"amount": amount,
"currency": (price.get("currency") or plans.PRICE_CURRENCY).upper(),
"interval": ((price.get("recurring") or {}).get("interval")
or plans.PRICE_INTERVAL),
}
_price_cache["at"] = now
_price_cache["prices"] = out
return out
def apply_live_prices(tiers: dict) -> dict:
"""Overlay Stripe's real amounts onto a plans.summary() `tiers` map, in place.
The single call the plan cards go through, so "what the card says" and "what
the checkout charges" cannot drift apart while Stripe is answering."""
live = live_prices()
for tier, info in tiers.items():
if tier in live:
info["price"] = live[tier]
return tiers
# --- hosted flows -----------------------------------------------------------
def ensure_customer(session, org: Organization) -> str:
"""The org's Stripe customer id, creating the customer on first use and
caching it on the subscription row so an org maps to exactly one customer for
its whole life."""
sub = plans.subscription_for(session, org.id)
if sub.stripe_customer_id:
return sub.stripe_customer_id
customer = _client().Customer.create(
name=org.name,
metadata={"org_id": str(org.id), "org_slug": org.slug},
)
sub.stripe_customer_id = customer["id"]
session.commit()
return customer["id"]
def create_checkout_session(session, org: Organization, tier: str,
success_url: str, cancel_url: str) -> str:
"""Start a subscription Checkout for `tier` and return the URL to redirect the
browser to. client_reference_id carries the org id so the completion webhook
can find the right tenant even before the customer is otherwise linked."""
if tier not in plans.PAID_TIERS:
raise Invalid(f"{tier!r} is not a purchasable plan.")
customer_id = ensure_customer(session, org)
checkout = _client().checkout.Session.create(
mode="subscription",
customer=customer_id,
client_reference_id=str(org.id),
line_items=[{"price": price_for_tier(tier), "quantity": 1}],
metadata={"org_id": str(org.id), "tier": tier},
success_url=success_url,
cancel_url=cancel_url,
)
return checkout["url"]
def create_portal_session(session, org: Organization, return_url: str) -> str:
"""Open the Stripe customer portal, where the customer changes plan, updates
their card, or cancels - all handled by Stripe, reflected back here by the
subscription.updated/deleted webhooks."""
customer_id = ensure_customer(session, org)
portal = _client().billing_portal.Session.create(
customer=customer_id, return_url=return_url,
)
return portal["url"]
# --- ending it --------------------------------------------------------------
#
# The portal above can already cancel, and it stays: it is Stripe's page, it is
# the only place a card can be changed, and it handles the cases this does not.
# What it cannot be is *inside* the product. Ending a subscription is the one
# thing a customer should never have to leave to do, and the two functions below
# are the two ways it ends: winding down at the end of a paid period (reversible
# until it lapses), and stopping outright because the whole account is going.
def set_cancel_at_period_end(session, org: Organization, cancel: bool) -> dict:
"""Ask Stripe to stop this subscription when the paid period ends, or take
that back. Returns the resulting {cancel_at_period_end, current_period_end}.
Deliberately not `Subscription.delete`: the customer has paid through a date,
so they keep the plan until it, and until then the decision is reversible -
`cancel=False` is Resume, the same call with the flag off. Stripe leaves such
a subscription `status = "active"` throughout, which is why the flag needs a
column of its own (see db_declarations.Subscription.cancel_at_period_end);
when the period finally ends Stripe sends subscription.deleted and the
existing handler drops the org to free.
The row is written from Stripe's response rather than from the argument, so
the page reflects what Stripe actually did instead of what we asked for, and
the webhook that follows finds the row already agreeing with it.
"""
sub = plans.subscription_for(session, org.id)
if not sub.stripe_subscription_id:
# Free, or a paid plan that has already lapsed. There is nothing at
# Stripe to modify, and pretending otherwise would report success for a
# cancellation that never happened.
raise Invalid("There is no paid plan on this organisation to cancel.")
updated = _client().Subscription.modify(
sub.stripe_subscription_id, cancel_at_period_end=bool(cancel),
)
sub.cancel_at_period_end = bool(updated.get("cancel_at_period_end", cancel))
sub.current_period_end = (_period_end(updated.get("current_period_end"))
or sub.current_period_end)
session.commit()
return {"cancel_at_period_end": sub.cancel_at_period_end,
"current_period_end": (sub.current_period_end.isoformat()
if sub.current_period_end else None)}
def _is_missing(exc) -> bool:
"""Whether a Stripe error is "that object is already gone". Treated as
success everywhere below: closing an account twice, or closing one whose
subscription was cancelled from the Stripe dashboard yesterday, must reach
the same end state rather than failing on the second attempt."""
return getattr(exc, "code", None) == "resource_missing"
def cancel_everything(session, org: Organization) -> None:
"""End every paid thing this organisation has at Stripe. Irreversible.
The other ending: the account is being closed, so there is no period to run
out and nothing to resume. Deleting the Customer is what does it - Stripe
cancels its subscriptions and detaches its payment methods on the way - and
it is the honest answer to "delete my account", since a cancelled
subscription would otherwise leave the customer's card on file forever. The
invoices survive it, which is what tax records require.
The subscription is cancelled explicitly first rather than left to the
Customer delete, so the state Stripe ends in is one this code asked for and
not a side effect. Both steps tolerate the object already being gone.
A no-op when billing is not configured on this deployment, or when the org
never reached a checkout - a free tenant has no Stripe customer to delete,
and closing it must not require a Stripe key to be set.
**Call this before you delete the tenant, not after.** If cancelling fails
the close should abort and the customer keeps their account; the other way
round, a Stripe outage leaves somebody with no account and a live
subscription still charging them.
"""
sub = plans.subscription_for(session, org.id)
if not is_enabled() or not sub.stripe_customer_id:
return
stripe = _client()
if sub.stripe_subscription_id:
try:
stripe.Subscription.delete(sub.stripe_subscription_id)
except Exception as e:
if not _is_missing(e):
raise
try:
stripe.Customer.delete(sub.stripe_customer_id)
except Exception as e:
if not _is_missing(e):
raise
logger.info("Cancelled every Stripe subscription for org %r.", org.slug)
# --- webhook: the source of truth folds back in -----------------------------
def construct_event(payload: bytes, sig_header: str):
"""Verify a webhook payload against STRIPE_WEBHOOK_SECRET and return the
parsed event, or raise Invalid. The signature is what lets an unauthenticated,
session-less endpoint trust that this really is Stripe and not a forged POST
trying to promote an org for free."""
secret = settings.Config.STRIPE_WEBHOOK_SECRET
if not secret:
raise Invalid("Billing webhook is not configured on this deployment.")
import stripe
try:
return stripe.Webhook.construct_event(payload, sig_header, secret)
except Exception as e:
# A bad signature, malformed body, or wrong secret all land here.
raise Invalid("Invalid webhook signature.") from e
def _period_end(ts) -> datetime | None:
"""A Stripe unix timestamp as a naive UTC datetime, matching the DateTime
columns everywhere else in this app. `utcfromtimestamp` is deprecated, so
this converts explicitly and then drops the tzinfo rather than storing an
aware value in a column the rest of the schema reads as naive."""
if not ts:
return None
return datetime.fromtimestamp(ts, tz=timezone.utc).replace(tzinfo=None)
def _sub_by_customer(session, customer_id: str) -> Subscription | None:
if not customer_id:
return None
return (session.query(Subscription)
.filter(Subscription.stripe_customer_id == customer_id)
.one_or_none())
def _sub_by_stripe_id(session, stripe_sub_id: str) -> Subscription | None:
if not stripe_sub_id:
return None
return (session.query(Subscription)
.filter(Subscription.stripe_subscription_id == stripe_sub_id)
.one_or_none())
def sync_from_event(session, event) -> bool:
"""Apply a verified Stripe event to the local Subscription row.
Idempotent: it resolves the org's row (by client_reference_id, then Stripe
customer, then subscription id) and writes the resulting state, so Stripe's
at-least-once retries and out-of-order delivery converge to the same row
rather than duplicating anything. Returns True if a row was updated, False
for an event we do not act on (so the endpoint can still 200 it - an
unhandled event is not an error).
"""
etype = event["type"]
obj = event["data"]["object"]
if etype == "checkout.session.completed":
org_id = obj.get("client_reference_id") or (obj.get("metadata") or {}).get("org_id")
sub = None
if org_id:
sub = plans.subscription_for(session, int(org_id))
sub = sub or _sub_by_customer(session, obj.get("customer"))
if sub is None:
logger.warning("checkout.session.completed for an unknown org: %s", obj.get("id"))
return False
sub.stripe_customer_id = obj.get("customer") or sub.stripe_customer_id
sub.stripe_subscription_id = obj.get("subscription") or sub.stripe_subscription_id
tier = (obj.get("metadata") or {}).get("tier")
if tier in plans.PAID_TIERS:
sub.tier = tier
sub.status = "active"
session.commit()
return True
if etype in ("customer.subscription.updated", "customer.subscription.created"):
sub = (_sub_by_stripe_id(session, obj.get("id"))
or _sub_by_customer(session, obj.get("customer")))
if sub is None:
logger.warning("subscription.updated for an unknown customer: %s", obj.get("customer"))
return False
sub.stripe_subscription_id = obj.get("id") or sub.stripe_subscription_id
price_id = _price_from_subscription(obj)
tier = tier_for_price(price_id) if price_id else None
if tier:
sub.tier = tier
sub.stripe_price_id = price_id
sub.status = obj.get("status") or sub.status
sub.current_period_end = _period_end(obj.get("current_period_end"))
# Whether it is winding down. Read from the event rather than only set by
# our own cancel call, so a cancellation taken in the Stripe portal or
# the Stripe dashboard shows up on the billing page too.
sub.cancel_at_period_end = bool(obj.get("cancel_at_period_end"))
session.commit()
return True
if etype == "customer.subscription.deleted":
sub = (_sub_by_stripe_id(session, obj.get("id"))
or _sub_by_customer(session, obj.get("customer")))
if sub is None:
return False
# The paid plan is gone: fall back to free, keep the customer id so a
# later re-subscribe reuses the same Stripe customer.
sub.tier = "free"
sub.status = "canceled"
sub.stripe_subscription_id = None
sub.stripe_price_id = None
sub.current_period_end = _period_end(obj.get("current_period_end"))
# The wind-down has finished, so the intention has nothing left to
# describe. Cleared rather than left true, or a later re-subscribe would
# start out claiming it was already cancelled.
sub.cancel_at_period_end = False
session.commit()
return True
if etype == "invoice.payment_failed":
sub = _sub_by_customer(session, obj.get("customer"))
if sub is None:
return False
# Mark it past_due; plans.effective_tier now treats the org as free until
# a successful payment flips it back to active via subscription.updated.
sub.status = "past_due"
session.commit()
return True
return False
def _price_from_subscription(sub_obj) -> str | None:
"""The price id out of a Stripe subscription object's first line item. Guards
every hop because a webhook payload's nested shape is not ours to assume."""
items = (sub_obj.get("items") or {}).get("data") or []
if not items:
return None
price = items[0].get("price") or {}
return price.get("id")