"""The PriceList aggregate — the rules, and nothing else.
The prices a group of customers pays in one currency. Its dated rows and its rules are written beside it
(infrastructure: one keyed write each, never the whole list), and the rules here say what may be written:
- a price row is NEVER edited: a different price needs a later valid_from, so a document priced last year
re-prices identically. A DRAFT row has never priced anything and may be discarded; a live one may not;
- a published list only takes prices (and rules) dated AFTER today: anything earlier would re-price a
document that already has a number;
- rows and rules stay DRAFT until published; only live ones price anything, and a price is derived on read.
"""
import re
from dataclasses import dataclass, field
from datetime import date
from decimal import Decimal
from typing import Iterable, Optional
from erp_platform.money import minor_unit
from brick_product.domain.errors import RuleBroken
from brick_product.domain.events import (
CustomerAssigned,
PriceListOpened,
PriceListPublished,
PriceListRetired,
PricesChanged,
PricesChecked,
PricesSet,
)
from brick_product.domain.pricing import ROUNDINGS, PriceRow, PriceRule, rate
from brick_product.domain.values import PriceListState, RowState, RuleMethod, RuleScope
LIST_CODE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]{0,63}$")
# ISO 4217 alphabetic codes in current use, with the funds codes a ledger can meet (XXX, XTS and metals are not
# currencies a customer is priced in).
ISO_4217 = frozenset("""AED AFN ALL AMD ANG AOA ARS AUD AWG AZN BAM BBD BDT BGN BHD BIF BMD BND BOB BOV BRL BSD BTN BWP BYN
BZD CAD CDF CHE CHF CHW CLF CLP CNY COP COU CRC CUP CVE CZK DJF DKK DOP DZD EGP ERN ETB EUR FJD FKP GBP GEL GHS GIP GMD
GNF GTQ GYD HKD HNL HTG HUF IDR ILS INR IQD IRR ISK JMD JOD JPY KES KGS KHR KMF KPW KRW KWD KYD KZT LAK LBP LKR LRD LSL
LYD MAD MDL MGA MKD MMK MNT MOP MRU MUR MVR MWK MXN MXV MYR MZN NAD NGN NIO NOK NPR NZD OMR PAB PEN PGK PHP PKR PLN PYG
QAR RON RSD RUB RWF SAR SBD SCR SDG SEK SGD SHP SLE SOS SRD SSP STN SVC SYP SZL THB TJS TMT TND TOP TRY TTD TWD TZS UAH
UGX USD USN UYI UYU UYW UZS VED VES VND VUV WST XAF XCD XCG XOF XPF YER ZAR ZMW ZWG""".split())
RATE_LIMIT = Decimal("1e14") # NUMERIC(19,4): fifteen digits before the point
QTY_LIMIT = Decimal("1e12")
def valid_list_code(code: str) -> str:
code = (code or "").strip()
if not LIST_CODE.match(code):
raise RuleBroken("a price list code is up to 64 letters, digits and . _ / - characters, e.g. LIST-GBP")
return code
def valid_currency(code: str) -> str:
code = (code or "").strip()
if code not in ISO_4217:
raise RuleBroken(f"a price list is in one currency, given as its ISO 4217 code (GBP, EUR, JPY …), and '{code}' "
"is not one")
return code
def default_rounding(currency: str) -> str:
"""The step a COMPUTED price is rounded to when the list says nothing: the currency's own smallest unit (0.01 for
GBP, 1.00 for JPY, 0.001 for KWD)."""
unit = minor_unit(currency)
return {0: "1.00", 1: "0.10", 2: "0.01", 3: "0.001"}.get(unit, "none")
def check_rounding(rounding: str, currency: str) -> None:
"""A list rounds to a step no finer than its currency's minor unit; x.99 needs a currency with cents."""
if rounding not in ROUNDINGS:
raise RuleBroken(f"rounding is one of {', '.join(ROUNDINGS)} (x.99 rounds to the nearest whole unit, less 1p)")
unit = minor_unit(currency)
if rounding == "x.99" and unit < 2:
raise RuleBroken(f"x.99 rounding needs a currency with hundredths; {currency} has {unit} decimal places")
if rounding not in ("none", "x.99") and Decimal(rounding) < Decimal(1).scaleb(-unit):
raise RuleBroken(f"{currency} has {unit} decimal places: a price cannot be rounded to {rounding}")
@dataclass(frozen=True)
class RowIn:
"""A price a caller asks to write: one product's price on one list from a date."""
product_id: str
valid_from: date
unit_price: Decimal
valid_to: Optional[date] = None
unit_code: str = "" # the unit the price is per (the use case fills in the product's sales unit)
pricing_quantity: int = 1 # v4: the price is per this many of unit_code
def checked(self) -> "RowIn":
if isinstance(self.pricing_quantity, bool) or not isinstance(self.pricing_quantity, int) or not 1 <= self.pricing_quantity <= 100000:
raise RuleBroken(f"{self.product_id}: a pricing quantity is a whole number from 1 to 100000")
if not (Decimal(0) <= self.unit_price < RATE_LIMIT):
raise RuleBroken(f"{self.product_id}: a price is zero or more")
if self.valid_to is not None and self.valid_to < self.valid_from:
raise RuleBroken(f"{self.product_id}: valid_to {self.valid_to} is before valid_from {self.valid_from}")
return self
def same_as(self, row: PriceRow) -> bool:
q = Decimal("0.0001")
return (self.unit_price.quantize(q) == row.unit_price.quantize(q) and self.valid_to == row.valid_to
and self.pricing_quantity == (row.pricing_quantity or 1))
def row(self, price_list_id: str) -> PriceRow:
return PriceRow(price_list_id, self.product_id, self.valid_from, self.unit_price, self.valid_to, self.unit_code,
self.pricing_quantity)
@dataclass(frozen=True)
class RuleIn:
"""A rule a caller asks to write: a price derived from another, in the list's stated order (seq)."""
seq: int
applies_to: str = RuleScope.ALL.value
target: str = ""
min_quantity: Decimal = Decimal(0)
valid_from: Optional[date] = None
valid_to: Optional[date] = None
method: str = RuleMethod.DISCOUNT_PCT.value
discount_pct: Decimal = Decimal(0)
fixed_price: Optional[Decimal] = None
surcharge_amount: Decimal = Decimal(0)
base_list: str = ""
unit_code: str = "" # the unit a fixed price or surcharge is stated in (the use case fills in a product's)
pricing_quantity: int = 1 # v4: the quantity a fixed price or surcharge is per
def checked(self, list_id: str) -> "RuleIn":
if isinstance(self.pricing_quantity, bool) or not isinstance(self.pricing_quantity, int) or not 1 <= self.pricing_quantity <= 100000:
raise RuleBroken(f"rule {self.seq}: a pricing quantity is a whole number from 1 to 100000")
if self.seq < 1:
raise RuleBroken("a rule's place in the list (seq) starts at 1")
if self.applies_to not in {s.value for s in RuleScope}:
raise RuleBroken(f"rule {self.seq}: applies_to is product, category or all")
if self.applies_to != RuleScope.ALL.value and not self.target:
raise RuleBroken(f"rule {self.seq}: name the {self.applies_to} it applies to")
if self.method not in {m.value for m in RuleMethod}:
raise RuleBroken(f"rule {self.seq}: method is fixed, discount_pct or formula")
if self.method == RuleMethod.FIXED.value and self.fixed_price is None:
raise RuleBroken(f"rule {self.seq}: a fixed rule needs its fixed_price")
if (self.method == RuleMethod.FIXED.value or (self.method == RuleMethod.FORMULA.value and self.surcharge_amount)) \
and not self.unit_code:
raise RuleBroken(f"rule {self.seq}: a fixed price or a surcharge is stated per unit — say which (unit_code); it "
"is stored with the rule and never re-read from the product")
if self.fixed_price is not None and not (Decimal(0) <= self.fixed_price < RATE_LIMIT):
raise RuleBroken(f"rule {self.seq}: a price is zero or more")
if not (Decimal(0) <= self.discount_pct <= Decimal(100)):
raise RuleBroken(f"rule {self.seq}: a discount is between 0 and 100 percent")
if not (Decimal(0) <= self.min_quantity < QTY_LIMIT):
raise RuleBroken(f"rule {self.seq}: the quantity from which it applies is zero or more")
if abs(self.surcharge_amount) >= RATE_LIMIT:
raise RuleBroken(f"rule {self.seq}: the surcharge is too large")
if self.valid_from and self.valid_to and self.valid_to < self.valid_from:
raise RuleBroken(f"rule {self.seq}: valid_to is before valid_from")
if self.base_list == list_id:
raise RuleBroken(f"rule {self.seq}: a list cannot be its own base — leave base_list empty to start from "
"this list's own prices")
return self
def same_as(self, r: PriceRule) -> bool:
return (self.applies_to == r.applies_to and (self.target or "") == (r.target or "")
and self.min_quantity == (r.min_quantity or 0) and self.valid_from == r.valid_from
and self.valid_to == r.valid_to and self.method == r.method
and self.discount_pct == (r.discount_pct or 0)
and (self.fixed_price if self.fixed_price is None else self.fixed_price.quantize(Decimal("0.0001")))
== (r.fixed_price if r.fixed_price is None else r.fixed_price.quantize(Decimal("0.0001")))
and self.surcharge_amount == (r.surcharge_amount or 0) and (self.base_list or "") == (r.base_list or "")
and (self.unit_code or "") == (r.unit_code or "") and self.pricing_quantity == (r.pricing_quantity or 1))
@dataclass
class RowsPlan:
"""What a write would do: the new rows and rules, rules to replace, and what was already there."""
new_rows: list
same_rows: list
new_rules: list
replaced_rules: list
same_rules: list
def plan_changes(price_list_id: str, rows: Iterable[RowIn], rules: Iterable[RuleIn], existing_rows: dict,
existing_rules: dict, *, today: date, published: bool) -> RowsPlan:
"""Decide a write against what is stored, before anything is written (an atomic refusal).
`existing_rows`: (product_id, unit_code, valid_from) → PriceRow; `existing_rules`: seq → (PriceRule, state).
Idempotent: what is already there, identically, changes nothing. Never edits a price."""
plan = RowsPlan([], [], [], [], [])
seen: dict = {}
for r in rows:
r = r.checked()
key = (r.product_id, r.unit_code, r.valid_from)
if key in seen:
if not r.same_as(seen[key].row(price_list_id)):
raise RuleBroken(f"{r.product_id} is given two different prices from {r.valid_from}")
continue
seen[key] = r
stored = existing_rows.get(key)
if stored is not None:
if r.same_as(stored):
plan.same_rows.append(r)
continue
raise RuleBroken(f"{r.product_id} already has a price of {rate(stored.unit_price)} from "
f"{r.valid_from} in {price_list_id}: a price is never edited — add the new price "
"from a later date")
if published and r.valid_from <= today:
raise RuleBroken(f"{r.product_id} from {r.valid_from}: a published list only takes prices from a later "
f"date than today ({today}) — a price already in force is never edited")
plan.new_rows.append(r)
seqs: set = set()
for rule in rules:
rule = rule.checked(price_list_id)
if rule.seq in seqs:
raise RuleBroken(f"rule {rule.seq} is given twice")
seqs.add(rule.seq)
stored = existing_rules.get(rule.seq)
if stored is not None:
old, state = stored
if rule.same_as(old):
plan.same_rules.append(rule)
continue
if state == RowState.LIVE.value or published:
raise RuleBroken(f"rule {rule.seq} of {price_list_id} is in force and is never edited — add a new "
"rule from a later date")
plan.replaced_rules.append(rule)
continue
if published and not (rule.valid_from and rule.valid_from > today):
raise RuleBroken(f"rule {rule.seq}: a published list only takes rules that start after today "
f"({today}) — set its valid_from")
plan.new_rules.append(rule)
return plan
@dataclass
class PriceList:
"""the prices a group of customers pays in one currency"""
price_list_id: str
name: str = ""
currency: str = ""
based_on: str = ""
prices_include_tax: bool = False
rounding: str = "0.01"
default_from: Optional[date] = None # the currency's default list from this date, until a later default_from
opened_by: str = ""
status: str = PriceListState.DRAFT.value
retired_on: Optional[date] = None # the business day it was retired: it still answers dates up to then
publish_notes: str = "" # JSON: what publishing flagged (a price far from its base), kept with the list
events: list = field(default_factory=list, repr=False, compare=False)
def _record(self, event):
self.events.append(event)
return event
# ── the lifecycle (module.yaml workflows) ──────────────────────────────────────────────────────────────
@classmethod
def open_price_list(cls, price_list_id: str, *, by: str = "", currency: str = "", name: str = "",
based_on: str = "", prices_include_tax: bool = False, rounding: str = "",
default_from: Optional[date] = None) -> "PriceList":
"""Open a price list: none → draft (PRD-05). Gate: a currency (ISO 4217); rounding no finer than the
currency's minor unit. Effect: based on another list, or standalone (that the base exists, in the same
currency and with the same VAT basis, is the use case's to check). `default_from`: the date it becomes the
currency's default list — checked to be after today when it is published."""
it = cls(valid_list_code(price_list_id), name=(name or "").strip() or price_list_id,
currency=valid_currency(currency), based_on=(based_on or "").strip(),
prices_include_tax=bool(prices_include_tax), rounding=rounding or default_rounding(currency),
default_from=default_from, opened_by=by)
check_rounding(it.rounding, it.currency)
if it.based_on == it.price_list_id:
raise RuleBroken("a list cannot be based on itself")
it._record(PriceListOpened(price_list_id=it.price_list_id, by=by))
return it
def set_prices(self, *, by: str = "") -> PricesSet:
"""Set prices and rules: draft → draft. Effect: dated rows; nothing visible to sales yet."""
if self.status != PriceListState.DRAFT:
raise RuleBroken(f"{self.price_list_id} is {self.status}: prices are set on a draft list "
"(a published list takes a scheduled change)")
return self._record(PricesSet(price_list_id=self.price_list_id, by=by))
def ai_price_check(self, mode: str, *, by: str = "") -> PricesChecked:
"""AI flags unusual prices (suggest): draft → draft. Gate: the price check is on. Effect: outliers are
listed for the pricing manager — nothing is edited."""
if self.status != PriceListState.DRAFT:
raise RuleBroken(f"{self.price_list_id} is {self.status}: the price check is for a draft list")
if mode == "off":
raise RuleBroken("the price check is switched off in the settings")
return self._record(PricesChecked(price_list_id=self.price_list_id, by=by))
def publish_price_list(self, *, by: str = "", authors: Iterable[str] = (), pending: int = 0,
unpublished_bases: Iterable[str] = (), second_person: bool = False,
backdated: Iterable[str] = (), today: Optional[date] = None,
default_clash: str = "", first_default: bool = False) -> Optional[PriceListPublished]:
"""Publish the price list: draft → published (PRD-07), or put a scheduled change live (published →
published, nothing pending changes nothing). Gate: every row and rule starts after today (nothing published changes a price for a date already past); a
default_from, if set, is after today — or today, for the first default list a currency ever has — and no other list
of the currency has the same; every list a rule is
based on is published; where the setting asks, the approver is not the author. Effect: sales prices from it
on each row's valid_from; from its default_from it is the currency's default list."""
if self.status == PriceListState.RETIRED:
raise RuleBroken(f"{self.price_list_id} is retired and cannot be published again")
was_published = self.status == PriceListState.PUBLISHED
if was_published and not pending:
return None # nothing waiting: a repeat changes nothing
late = list(backdated)
if late:
raise RuleBroken(f"publishing {self.price_list_id} now would change prices for dates already past: "
+ "; ".join(late[:8]) + (f" … and {len(late) - 8} more" if len(late) > 8 else "")
+ ". A price already in force is never edited: discard these drafts and schedule them "
"from a later date.")
if not was_published and self.default_from is not None:
# the first default a currency ever has may start today: nothing was priced from a default before it
if today is not None and (self.default_from < today if first_default else self.default_from <= today):
raise RuleBroken(f"{self.price_list_id} would be the default {self.currency} list from "
f"{self.default_from}, which is not after today ({today}): a default starts tomorrow "
"or later, or the price on a past date would change"
+ ("" if first_default else " (only the first default a currency has may start today)"))
if default_clash:
raise RuleBroken(f"{default_clash} is already the default {self.currency} list from "
f"{self.default_from}: a currency has one default on any date")
bases = sorted(set(unpublished_bases))
if bases:
raise RuleBroken(f"{self.price_list_id} has a rule based on {', '.join(bases)}, which is not published: "
"publish it first")
me = (by or "").strip().lower()
if second_person and me and me in {a.strip().lower() for a in authors if a}:
raise RuleBroken(f"a second person must publish {self.price_list_id}: you prepared it, and the "
"settings ask for someone else to approve it")
self.status = PriceListState.PUBLISHED.value
return self._record(PriceListPublished(price_list_id=self.price_list_id, by=by))
def change_prices(self, *, by: str = "") -> PricesChanged:
"""Schedule a price change: published → published. Effect: new rows with a later valid_from, waiting as
a draft change until published; old rows are never edited."""
if self.status != PriceListState.PUBLISHED:
raise RuleBroken(f"{self.price_list_id} is {self.status}: a change is scheduled on a published list")
return self._record(PricesChanged(price_list_id=self.price_list_id, by=by))
def assign_customer(self, bp_id: int, start: date, today: date, *, by: str = "") -> CustomerAssigned:
"""Assign customers to the list: published → published. Gate: the assignment starts after today. Effect:
their prices in this currency resolve from this list from that date (PRD-08)."""
if self.status != PriceListState.PUBLISHED:
raise RuleBroken(f"{self.price_list_id} is {self.status}: customers are priced from a published list")
if start <= today:
raise RuleBroken(f"an assignment starts tomorrow or later, not {start}: starting on or before today "
f"({today}) would change a price already given")
return self._record(CustomerAssigned(price_list_id=self.price_list_id, bp_id=bp_id, by=by))
def retire_price_list(self, today: date, *, by: str = "", used_by: Iterable[str] = (),
default_to: Optional[date] = None, default_open: bool = False) -> Optional[PriceListRetired]:
"""Retire the price list: published → retired. Gate: not the default of its currency today or later, and not
the base of another list or rule. Effect: assigned customers fall back to the default list; prices for past
dates still resolve from it (it answers every date up to today)."""
if self.status == PriceListState.RETIRED:
return None # already retired: a repeat changes nothing
if self.status != PriceListState.PUBLISHED:
raise RuleBroken(f"a {self.status} price list cannot be retired: only a published one")
if default_open:
raise RuleBroken(f"{self.price_list_id} is the default {self.currency} list from {self.default_from}"
+ (f" until {default_to}" if default_to else " with no later default")
+ ", today or later: publish the list that replaces it first, or customers with no list "
"of their own would have no price")
users = sorted(set(used_by))
if users:
raise RuleBroken(f"{self.price_list_id} is still the base of {', '.join(users)}: retiring it would leave "
"them without the prices they start from — retire or re-base those first")
self.status = PriceListState.RETIRED.value
self.retired_on = today
return self._record(PriceListRetired(price_list_id=self.price_list_id, by=by))