subscriptions.usage-tiered

Imports name this capability’s declared dependencies, which fune builds next to it in your project; each one links to its page.

from typing import List, Sequence

from .money_amount import assert_same_currency, money  ← from money.amount ^1.0.0 · built alongside by fune
from .money_sum import sum_money  ← from money.sum ^1.0.0 · built alongside by fune
from .subscriptions_usage_tiered_types import TierCharge, TierMode, UsageCharge, UsageTier


def _charge(tier: int, quantity: int, t: UsageTier) -> TierCharge:
    amount = money(quantity * t.unit_price.minor + t.flat_fee.minor, t.unit_price.currency)
    return TierCharge(tier=tier, quantity=quantity, unit_price=t.unit_price, flat_fee=t.flat_fee, amount=amount)


def price_usage(quantity: int, tiers: Sequence[UsageTier], mode: TierMode) -> UsageCharge:
    """Price ``quantity`` units against ascending tiers. Graduated prices each
    unit in its own tier and charges the flat fee of every tier reached; volume
    prices every unit at the tier the quantity falls in, plus that tier's flat
    fee. The first tier is always reached, so its flat fee applies at zero.
    """
    if isinstance(quantity, bool) or not isinstance(quantity, int) or quantity < 0:
        raise ValueError("quantity must be a whole number of 0 or more, received %r" % (quantity,))
    if mode not in ("graduated", "volume"):
        raise ValueError('unknown tier mode "%s": expected graduated or volume' % (mode,))
    if len(tiers) == 0:
        raise ValueError("tiered pricing needs at least one tier")
    currency = tiers[0].unit_price.currency
    previous = 0
    for i, t in enumerate(tiers):
        assert_same_currency(tiers[0].unit_price, t.unit_price)
        assert_same_currency(tiers[0].unit_price, t.flat_fee)
        if t.up_to is None:
            if i != len(tiers) - 1:
                raise ValueError("only the last tier may be open-ended (upTo null)")
        else:
            if isinstance(t.up_to, bool) or not isinstance(t.up_to, int) or t.up_to <= previous:
                raise ValueError(
                    "tier upTo values must be positive and strictly increasing, received %r after %d"
                    % (t.up_to, previous)
                )
            previous = t.up_to
    last = tiers[-1]
    if last.up_to is not None and quantity > last.up_to:
        raise ValueError("quantity %d exceeds the last tier, which ends at %d" % (quantity, last.up_to))

    lines: List[TierCharge] = []
    if mode == "volume":
        index = next(i for i, t in enumerate(tiers) if t.up_to is None or quantity <= t.up_to)
        lines.append(_charge(index + 1, quantity, tiers[index]))
    else:
        floor = 0
        for i, t in enumerate(tiers):
            # Tier 1 is always reached; a later tier only once usage passes the
            # previous tier's last unit.
            if i > 0 and quantity <= floor:
                break
            top = quantity if t.up_to is None else min(quantity, t.up_to)
            lines.append(_charge(i + 1, top - floor, t))
            if t.up_to is None:
                break
            floor = t.up_to
    return UsageCharge(lines=lines, total=sum_money([line.amount for line in lines], currency))