from typing import List, Sequence from .money_amount import assert_same_currency, money from .money_sum import sum_money 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))