"""Subscription tiers and the entitlements each one carries.
The one place the paywall's rules are written down. Everything else - the
endpoints, the billing page, the webhook handler - reads a tier and asks this
module "is this allowed?", exactly as auth.py's ROLE_SCOPES is the one place a
role's powers are defined. Changing what Silver includes is an edit to TIERS
here, not a migration and not a hunt through endpoints.
Two kinds of entitlement:
limits a cap on a countable resource. None means unlimited. enforce_limit
counts current usage and refuses the create that would exceed the
cap.
features a boolean gate on a capability. require_feature refuses when the
plan does not include it.
And, since a plan nobody can price is a plan nobody buys, what each tier costs -
see PRICE_CURRENCY below for why the number lives here as well as in Stripe.
A refusal is PaymentRequired, not Forbidden: it is fixed by upgrading, so the
façade answers 402 and the message points at the upgrade. Nothing here imports
db_session or raises HTTP - it takes a session and raises a domain error, like
the rest of app/core.
**Rename all of this.** The resources and features below are the skeleton's own,
because a paywall with nothing behind it is decoration and would not survive
contact with a real product. `subcategory` is the tree's leaf and `api_tokens` is
machine access; both are worked examples, and both are one edit away from being
whatever your product sells.
"""
import logging
from sqlalchemy import func
from sqlalchemy.exc import IntegrityError
import settings
from db.db_declarations import Membership, Subcategory, Subscription, User
from .context import Invalid, PaymentRequired
logger = logging.getLogger(__name__)
# The resource keys enforce_limit understands, and the feature keys
# require_feature understands. Named here so a typo at a call site is caught
# rather than silently treated as "unlimited"/"off".
RESOURCES = ("subcategory", "seat")
FEATURES = ("api_tokens",)
# What a plan costs, in the currency every tier is priced in. `price` below is an
# amount in **minor units** - Stripe's own convention, where 2900 is CHF 29.00 -
# so money is never held in a float, and the same integer can be compared with
# what Stripe reports without a rounding argument.
#
# Stripe owns the amount that is actually charged. These are the *declared*
# prices: what the app renders with no network, in CI, and on a deployment with
# no Stripe key (int, which is deployed without STRIPE_* on purpose).
# billing.live_prices() reads the real amounts off the configured Prices and
# overlays them whenever Stripe is reachable, so a card can only be stale, never
# contradicted by the checkout it leads to.
#
# Change a price in the Stripe dashboard and change it here in the same sitting.
# This is the figure a customer is shown for the seconds Stripe is slow, and a
# fallback that quotes the wrong plan's price is worse than one that quotes none.
PRICE_CURRENCY = "CHF"
PRICE_INTERVAL = "month"
# The whole tier definition. `limits` maps a resource to its cap (None =
# unlimited); a resource absent from the map is unlimited too. `features` is the
# set of feature keys the tier includes. `price` is per organisation per
# PRICE_INTERVAL - flat, not per seat, which is what checkout's quantity of 1
# already buys. Ordered cheapest-first in TIER_ORDER so an upgrade/downgrade
# direction can be reasoned about.
TIERS = {
"free": {
"label": "Free",
# Zero, not None: "this costs nothing" is a price, and a card saying
# CHF 0 answers the question a card saying nothing leaves open.
"price": 0,
"limits": {"subcategory": 3, "seat": 2},
"features": set(),
},
"silver": {
"label": "Silver",
"price": 900,
"limits": {"subcategory": 25, "seat": 10},
"features": {"api_tokens"},
},
"gold": {
"label": "Gold",
"price": 2900,
"limits": {"subcategory": None, "seat": None},
"features": {"api_tokens"},
},
}
TIER_ORDER = ("free", "silver", "gold")
# The tiers a customer can buy (free is the default, not a purchase). Each maps
# to a settings key holding its Stripe price id - resolved in billing.py so this
# module stays free of Stripe.
PAID_TIERS = ("silver", "gold")
def is_tier(tier: str) -> bool:
return tier in TIERS
def label(tier: str) -> str:
return TIERS.get(tier, {}).get("label", tier.title())
def price(tier: str) -> dict:
"""What `tier` costs, in the shape a page can render: an amount in minor
units, the currency it is in, and what one payment of it buys.
An unknown tier prices as free rather than raising. label() takes the same
line: a price is a description, and a page must still draw itself when the
row in front of it names a tier this build has never heard of."""
return {
"amount": TIERS.get(tier, TIERS["free"])["price"],
"currency": PRICE_CURRENCY,
"interval": PRICE_INTERVAL,
}
# --- the org's subscription -------------------------------------------------
def subscription_for(session, org_id: int) -> Subscription:
"""The organisation's subscription row, created free if it is missing.
Every org gets one at sign-up, so this normally just fetches; the create is a
belt-and-braces default so a read never returns None and callers need no null
handling. It is also what makes the paywall retrofittable: an org that
predates billing gets its free row on first read rather than needing a
backfill nobody remembers to run.
"""
sub = (session.query(Subscription)
.filter(Subscription.org_id == org_id)
.one_or_none())
if sub is None:
sub = Subscription(org_id=org_id, tier="free", status="active")
session.add(sub)
try:
session.commit()
except IntegrityError:
# Another path (a concurrent request, or a caller that seeded the row
# in a different session) created the org's subscription between the
# query above and this commit. The uq_subscription_org constraint
# rejects the second insert; roll the failed transaction back - so the
# session is not left in a state where every later statement raises
# PendingRollbackError - and return the row that already exists.
session.rollback()
sub = (session.query(Subscription)
.filter(Subscription.org_id == org_id)
.one())
return sub
# The statuses that actually entitle a paid tier. A subscription that is
# past_due/canceled/incomplete falls back to free access - the paid features
# switch off the moment Stripe says the money stopped, without deleting the row
# (so the billing page can still explain what happened and offer to fix it).
ENTITLING_STATUSES = frozenset({"active", "trialing"})
def effective_tier(sub: Subscription) -> str:
"""The tier whose entitlements actually apply. A paid tier only counts while
its subscription is active/trialing; otherwise the org is treated as free."""
if sub.tier in PAID_TIERS and sub.status not in ENTITLING_STATUSES:
return "free"
return sub.tier
def tier_for_org(session, org_id: int) -> str:
return effective_tier(subscription_for(session, org_id))
# --- the accounts that are not billed ---------------------------------------
def _exempt_emails() -> frozenset:
"""The configured allowlist, normalised. Read per call rather than cached at
import so a test (and a redeploy) can change it without reloading the
module."""
raw = settings.Config.BILLING_EXEMPT_EMAILS or ""
return frozenset(e.strip().lower() for e in raw.split(",") if e.strip())
def is_exempt(session, org_id: int) -> bool:
"""Whether this organisation is not billed at all.
True when an allowlisted account holds the **owner** role in it. Owner and
not merely a membership on purpose: being invited into a customer's
organisation - to look at something, to help - must never switch that
customer's paywall off. It is also why the exemption is keyed on the org
rather than on the caller: a limit is a property of the tenant, so an org
cannot be exempt for one of its members and capped for another.
An empty allowlist short-circuits before touching the database, so every
deployment that does not configure one pays nothing for this - not a query,
not a behaviour change.
"""
emails = _exempt_emails()
if not emails:
return False
return (session.query(Membership.id)
.join(User, User.id == Membership.user_id)
.filter(Membership.org_id == org_id,
Membership.role == "owner",
func.lower(User.email).in_(emails))
.first() is not None)
# --- limit lookups and counting ---------------------------------------------
def limit(tier: str, resource: str):
"""The cap on `resource` for `tier`, or None for unlimited."""
if resource not in RESOURCES:
raise Invalid(f"Unknown resource {resource!r}.")
return TIERS.get(tier, TIERS["free"])["limits"].get(resource)
def has_feature(tier: str, feature: str) -> bool:
if feature not in FEATURES:
raise Invalid(f"Unknown feature {feature!r}.")
return feature in TIERS.get(tier, TIERS["free"])["features"]
def effective_limit(session, org_id: int, resource: str):
"""The org's real cap on `resource`: its tier cap, or None (unlimited) if it
is not billed. The one lookup every gate and every meter goes through, so a
refusal and the "3 of 5" it is explained by can never disagree.
`resource` is validated even for an exempt org - a typo at a call site must
still fail loudly rather than read as unlimited.
"""
cap = limit(tier_for_org(session, org_id), resource)
if is_exempt(session, org_id):
return None
return cap
def _count(session, org_id: int, resource: str) -> int:
"""Current usage of a countable resource in one organisation.
One function, so a meter and the refusal beside it count the same rows. Add a
resource here and to RESOURCES and to every tier's `limits` in the same edit;
a key in one and not the others is the failure this module is shaped to make
loud rather than silent.
"""
if resource == "subcategory":
return (session.query(Subcategory)
.filter(Subcategory.org_id == org_id).count())
if resource == "seat":
return (session.query(Membership)
.filter(Membership.org_id == org_id).count())
raise Invalid(f"Unknown resource {resource!r}.")
def usage(session, org_id: int) -> dict:
"""Current usage of every countable resource, for the billing page's
"3 of 5" meters."""
return {r: _count(session, org_id, r) for r in RESOURCES}
# --- enforcement (the paywall) ----------------------------------------------
def enforce_limit(session, org_id: int, resource: str) -> None:
"""Refuse the create that would push `resource` past the org's tier cap.
Called at a create boundary *before* the row is added, so the check is on
the count that would result (current + 1). Unlimited tiers - and orgs that
are not billed at all - have cap None and pass straight through. A hit cap
raises PaymentRequired, which the façade turns into a 402 with this upsell
message.
"""
cap = effective_limit(session, org_id, resource)
if cap is None:
return
current = _count(session, org_id, resource)
if current >= cap:
# The plural the rest of the product uses, not the model's class name:
# this sentence is the upsell that sends people to the billing page, and
# the meter waiting for them there is labelled with the same word.
noun = {"subcategory": "pages", "seat": "members"}[resource]
raise PaymentRequired(
f"Your {label(tier_for_org(session, org_id))} plan includes "
f"{cap} {noun}. Upgrade to add more."
)
def require_feature(session, org_id: int, feature: str) -> None:
"""Refuse a feature the org's tier does not include (402 upsell). An org that
is not billed has every feature, including any added after this was written -
which is why the exemption sits here rather than as a fourth entry in
TIERS."""
tier = tier_for_org(session, org_id)
if has_feature(tier, feature):
return
if is_exempt(session, org_id):
return
raise PaymentRequired(
f"{feature_label(feature)} is not part of your {label(tier)} plan. "
"Upgrade to use it."
)
# What a feature is called when it is said out loud. Used by the refusal above
# and by the landing page's price list, so the thing a stranger is told they are
# buying is named by the same table that gates it - a plan card that invents its
# own wording is how a page ends up advertising a feature the tier does not have.
FEATURE_LABELS = {"api_tokens": "API tokens"}
def feature_label(feature: str) -> str:
return FEATURE_LABELS.get(feature, feature)
def catalogue() -> dict:
"""The tiers as a price list, with no organisation in the picture.
`summary()` below answers "what does *this* org have"; this answers "what is
on offer", which is the question a signed-out pricing page asks. It takes no
session because there is nobody to look up - a stranger reading the pricing
section is not a member of anything.
Same per-tier shape as `summary()["tiers"]` on purpose, so
`billing.apply_live_prices` overlays Stripe's real amounts onto either one
without knowing which it was handed.
"""
return {t: {"label": TIERS[t]["label"],
"price": price(t),
"limits": TIERS[t]["limits"],
"features": sorted(TIERS[t]["features"]),
# What the tier adds to the one below it, named. A price card
# lists the difference rather than the whole set - "everything in
# Silver, plus…" is how a customer reads a ladder - and deriving
# it here means a tier that gains a feature gains the bullet.
"adds": [feature_label(f) for f in sorted(
TIERS[t]["features"] - TIERS[_below(t)]["features"])],
"below_label": TIERS[_below(t)]["label"]}
for t in TIER_ORDER}
def _below(tier: str) -> str:
"""The tier one step cheaper, or `tier` itself at the bottom of the ladder -
so the cheapest tier's `adds` is its whole feature set (empty, for Free)
rather than a subtraction that has nothing to subtract from."""
i = TIER_ORDER.index(tier)
return TIER_ORDER[i - 1] if i else tier
def summary(session, org_id: int) -> dict:
"""Everything the billing UI needs about the org's plan in one call: the
effective tier, the raw subscription, per-resource usage-vs-limit, and which
features are on.
`limits` and `features` are what actually gates, so an exempt org reports
unlimited everything - the billing page would otherwise show a Free plan's
caps next to meters nothing is enforcing. `exempt` is carried separately so
the page can say why, and `tier_limits` keeps the tier's raw numbers visible.
"""
sub = subscription_for(session, org_id)
tier = effective_tier(sub)
exempt = is_exempt(session, org_id)
return {
"subscription": sub.to_dict(),
"effective_tier": tier,
"label": label(tier),
"exempt": exempt,
"limits": {r: (None if exempt else limit(tier, r)) for r in RESOURCES},
"tier_limits": {r: limit(tier, r) for r in RESOURCES},
"usage": usage(session, org_id),
"features": {f: (exempt or has_feature(tier, f)) for f in FEATURES},
"tiers": {t: {"label": TIERS[t]["label"],
"price": price(t),
"limits": TIERS[t]["limits"],
"features": sorted(TIERS[t]["features"])}
for t in TIER_ORDER},
}