dodil/erp-brick-purchasing
DODIL ERP brick: purchasing
src/brick_purchasing/domain/order.py
"""A purchase order: what we agreed to buy from a named supplier, at a price (PUR-04 to PUR-09, PUR-24, PUR-35, PUR-37, PUR-38).
A line is ordered in the supplier's pack (`unit_code`, with its exact conversion to the base unit copied at line time) and priced in
its ORDER PRICE UNIT (`pricing_unit_code`, its exact conversion FIXED when the line is priced: never re-read). Received, billed and
outstanding are derived from receipt and bill rows by the use case; nothing here stores them."""
import json
from dataclasses import dataclass, field
from datetime import date, datetime, timezone
from decimal import Decimal
from fractions import Fraction
from typing import Optional
from brick_purchasing.domain import approval
from brick_purchasing.domain.approval import Release, Strategy, identities
from brick_purchasing.domain.errors import RuleBroken
from brick_purchasing.domain.readiness import Gate, no, ok
from brick_purchasing.domain.values import (
ZERO,
as_dec,
base_of,
fits_money,
line_net,
priced_of,
)
@dataclass
class OrderLine:
purchase_order_id: str
line_no: int
product_id: str = ""
description: str = ""
quantity: Decimal = ZERO
unit_code: str = ""
unit_numerator: int = 1
unit_denominator: int = 1
quantity_base: Decimal = ZERO
base_unit_code: str = ""
pricing_unit_code: str = ""
pricing_numerator: int = 1
pricing_denominator: int = 1
pricing_conversion_source: str = "product" # product | supplier_price | unit
pricing_quantity: int = 1
unit_price: Decimal = ZERO
quantity_priced: Decimal = ZERO
net_amount: Decimal = ZERO
price_source: str = "manual" # supplier_price | manual | quote | agreement
tax_category: str = ""
expense_category: str = "general"
kind: str = "goods"
expected_on: Optional[date] = None
account_assignment: str = "stock"
cost_centre: str = ""
asset_id: str = ""
asset_class_code: str = ""
project_id: str = ""
company_code: str = ""
item_category: str = "standard"
agreement_id: str = ""
agreement_line_no: Optional[int] = None
ers: bool = False
order_multiple: Optional[Decimal] = None # the supplier price row's own multiple (checked instead of product's)
rounding_met: bool = True
rounding_reason: str = ""
rounding_fix: str = ""
quota_reason: str = "" # why the buyer took another supplier than the quota proposed
supplier_product_code: str = ""
@property
def exact_base(self) -> Fraction:
return base_of(self.quantity, self.unit_numerator, self.unit_denominator)
@property
def converts(self) -> bool:
return self.unit_numerator > 0 and self.unit_denominator > 0
@property
def pricing_converts(self) -> bool:
return self.pricing_numerator > 0 and self.pricing_denominator > 0
def reprice(self, currency: str, quantity=None) -> None:
"""Recompute quantity_base, quantity_priced and net from the line's FIXED conversions and price. A changed quantity keeps
the conversion (an amendment never re-reads the product), exact fractions, one rounding."""
if quantity is not None:
self.quantity = quantity
qb = self.exact_base
self.quantity_base = as_dec(qb)
qp = priced_of(qb, self.pricing_numerator, self.pricing_denominator)
self.quantity_priced = as_dec(qp)
self.net_amount = line_net(qp, self.unit_price, self.pricing_quantity, currency)
@dataclass
class OrderRevision:
purchase_order_id: str
revision: int
total: Decimal = ZERO
reason: str = ""
changed_by: str = ""
changed_at: Optional[datetime] = None
lines_json: str = "[]"
@dataclass
class OrderSend:
purchase_order_id: str
sent_at: datetime
sent_to: str = ""
sent_by: str = ""
revision: int = 1
@dataclass
class GateOverride:
purchase_order_id: str
revision: int
line_no: int
gate: str
failed_reason: str = ""
accepted_by_email: str = ""
accepted_by_sub: str = ""
accepted_on: Optional[date] = None
reason: str = ""
@dataclass
class OrderComponent:
purchase_order_id: str
line_no: int
component_no: int
product_id: str = ""
quantity_per: Decimal = ZERO
unit_code: str = ""
provided: Decimal = ZERO # the sum of provision moves, kept with the move refs
provision_move_refs: str = "[]"
allowance_reason: str = ""
@dataclass
class DeliveryCost:
purchase_order_id: str
line_no: int # 0 = the whole order
kind: str
seq: int
amount: Optional[Decimal] = None
percent: Optional[Decimal] = None
payee_bp_id: Optional[int] = None
currency: str = ""
tax_code: str = ""
@dataclass
class PurchaseOrder:
purchase_order_id: str
order_no: str = ""
revision: int = 1
bp_id: int = 0
currency: str = "GBP"
status: str = "draft"
buyer_email: str = ""
buyer_sub: str = ""
source_kind: str = "direct" # request | direct | rfq | agreement | schedule | odoo
source_id: str = ""
request_id: str = ""
rfq_id: str = ""
agreement_id: str = ""
release_no: Optional[int] = None
supplier_ref: str = ""
terms_id: str = ""
expected_on: Optional[date] = None
company_code: str = ""
confirmed_on: Optional[date] = None
approved_total: Decimal = ZERO
approved_by_email: str = ""
approved_by_sub: str = ""
approved_on: Optional[date] = None
submitted_on: Optional[date] = None
release_levels: int = 1
release_thresholds: str = "[]"
strategy_valid_from: Optional[date] = None
net_total: Decimal = ZERO
ship_to_bp_id: Optional[int] = None
ship_to_address: str = ""
sales_order_id: str = ""
so_line_no: Optional[int] = None
warehouse_id: str = ""
location_id: str = ""
cancel_reason: str = ""
close_reason: str = ""
lines: list = field(default_factory=list, repr=False)
revisions: list = field(default_factory=list, repr=False)
sends: list = field(default_factory=list, repr=False)
overrides: list = field(default_factory=list, repr=False)
components: list = field(default_factory=list, repr=False)
delivery_costs: list = field(default_factory=list, repr=False)
releases: list = field(default_factory=list, repr=False)
events: list = field(default_factory=list, repr=False)
# ── who ────────────────────────────────────────────────────────────────────────────────
@property
def owner_ids(self) -> frozenset:
return identities(self.buyer_email, self.buyer_sub)
def line(self, line_no: int) -> Optional[OrderLine]:
return next((x for x in self.lines if x.line_no == int(line_no)), None)
# ── money ──────────────────────────────────────────────────────────────────────────────
def recompute(self) -> None:
self.net_total = sum((x.net_amount for x in self.lines), ZERO)
if not fits_money(self.net_total) or not all(fits_money(x.net_amount) for x in self.lines):
raise RuleBroken("the order total does not fit an amount (at most 16 digits before the decimal point)")
@property
def total(self) -> Decimal:
return self.net_total
def covered(self) -> bool:
"""An approved request or approval covers the total."""
return self.approved_total >= self.net_total and self.approved_total > 0
def needs_approval(self, threshold: Decimal) -> bool:
"""No approval covers the total, and the total is above the owner's threshold (0 = every direct order)."""
return not self.covered() and self.net_total > threshold
# ── the draft ──────────────────────────────────────────────────────────────────────────
def set_lines(self, lines: list) -> None:
if self.status != "draft":
raise RuleBroken(f"an order that is {self.status} cannot be re-priced: amend it instead")
self.lines = lines
for x in lines:
x.purchase_order_id = self.purchase_order_id
self.recompute()
self.overrides = [o for o in self.overrides if self._override_still_valid(o)]
def _override_still_valid(self, o: GateOverride) -> bool:
x = self.line(o.line_no)
return x is not None and not x.rounding_met and x.rounding_reason == o.failed_reason
# ── approval ──────────────────────────────────────────────────────────────────────────
def submit(self, today: date, strategy: Strategy, threshold: Decimal) -> None:
if self.status != "draft":
raise RuleBroken(f"an order that is {self.status} cannot be sent for approval")
if not self.lines:
raise RuleBroken("an order with no lines cannot be sent for approval")
if not self.needs_approval(threshold):
raise RuleBroken("this order needs no approval: an approved request covers it, or it is within the owner's threshold — confirm it")
self.status, self.submitted_on = "to_approve", today
self.release_levels, self.strategy_valid_from = strategy.levels, strategy.valid_from
self.release_thresholds = json.dumps([str(t) for t in strategy.thresholds])
def approve_level(self, *, email, sub, limit, today) -> Release:
self._to_approve()
return approval.step(self, "order", self.purchase_order_id, self.revision, owner_ids=self.owner_ids, email=email, sub=sub,
limit=limit, needed=self.net_total, today=today, noun="order", final=False)
def approve(self, *, email, sub, limit, today) -> None:
self._to_approve()
approval.step(self, "order", self.purchase_order_id, self.revision, owner_ids=self.owner_ids, email=email, sub=sub,
limit=limit, needed=self.net_total, today=today, noun="order", final=True)
self.status, self.approved_on = "approved", today
self.approved_by_email, self.approved_by_sub = (email or "").lower(), (sub or "").lower()
self.approved_total = max(self.approved_total, self.net_total)
def reject(self, reason: str) -> None:
self._to_approve()
if not reason.strip():
raise RuleBroken("sending an order back needs a reason")
self.status = "draft"
self.releases = [r for r in self.releases if r.revision != self.revision]
self.close_reason = ""
def _to_approve(self) -> None:
if self.status != "to_approve":
raise RuleBroken(f"the order is {self.status}, not waiting for approval")
def waiting(self) -> str:
done = len([r for r in self.releases if r.revision == self.revision])
return approval.stage_note(self.release_levels, done) if self.status == "to_approve" else ""
# ── rounding override (PUR-37) ─────────────────────────────────────────────────────────
def accept_rounding(self, line_no: int, revision: int, *, email, sub, reason, today) -> GateOverride:
x = self.line(line_no)
if x is None:
raise RuleBroken(f"the order has no line {line_no}")
if x.rounding_met:
raise RuleBroken(f"line {line_no} meets its purchase rounding: there is nothing to accept")
if not (reason or "").strip():
raise RuleBroken("accepting an order off the pack rule needs a reason")
who = identities(email, sub)
if not who or approval.same_person(who, self.owner_ids):
from brick_purchasing.domain.errors import GateUnmet
raise GateUnmet([no("second_person", "you raised this order (same email or same sign-in)", "another budget holder")])
o = GateOverride(self.purchase_order_id, int(revision), int(line_no), "purchase_rounding", x.rounding_reason, (email or "").lower(),
(sub or "").lower(), today, reason.strip())
self.overrides = [p for p in self.overrides if (p.revision, p.line_no, p.gate) != (o.revision, o.line_no, o.gate)] + [o]
return o
def rounding_gate(self, x: OrderLine) -> Gate:
if x.rounding_met:
return ok("purchase_rounding")
if any(o.line_no == x.line_no and o.revision == self.revision and o.gate == "purchase_rounding" and o.failed_reason == x.rounding_reason
for o in self.overrides):
return ok("purchase_rounding")
return no("purchase_rounding", x.rounding_reason or f"line {x.line_no} is off its purchase rounding",
x.rounding_fix or "order the next valid quantity, or a second person accepts it with a reason")
# ── confirm ───────────────────────────────────────────────────────────────────────────
def core_gates(self, threshold: Decimal) -> list:
"""The gates that need no neighbour: authority, units, pricing units, rounding, components."""
gates = []
if not self.lines:
gates.append(no("lines", "the order has no lines", "add a line"))
gates.append(ok("authority") if not self.needs_approval(threshold) else
no("authority", f"the total {self.net_total} is above the order threshold and no approval covers it",
"send it for approval (PUR-09)"))
bad = [x.line_no for x in self.lines if not x.converts]
gates.append(ok("units_convert") if not bad else
no("units_convert", f"line {', '.join(map(str, bad))}: the unit has no exact conversion to the product's base unit",
"Product: add the unit with its conversion"))
bad = [x.line_no for x in self.lines if not x.pricing_converts]
gates.append(ok("pricing_unit_converts") if not bad else
no("pricing_unit_converts", f"line {', '.join(map(str, bad))}: the order price unit has no exact conversion to the "
"product's base unit", "Product: add the unit, or price the line in its order unit"))
for x in self.lines:
g = self.rounding_gate(x)
if not g.met:
gates.append(g)
if all(self.rounding_gate(x).met for x in self.lines):
gates.append(ok("purchase_rounding"))
bad = [x.line_no for x in self.lines if x.item_category == "subcontract" and not [c for c in self.components if c.line_no == x.line_no]]
gates.append(ok("components_listed") if not bad else
no("components_listed", f"line {', '.join(map(str, bad))} is a subcontract line with no components",
"list what we provide to the supplier"))
return gates
def confirm(self, today: date, order_no: str) -> None:
if self.status not in ("draft", "approved"):
raise RuleBroken(f"an order that is {self.status} cannot be confirmed")
self.order_no = self.order_no or order_no
self.status, self.confirmed_on = "confirmed", self.confirmed_on or today
if not self.revisions:
self.snapshot("confirmed", "", today)
self.events.append(("order_confirmed", self.purchase_order_id))
def snapshot(self, reason: str, by: str, today) -> None:
self.revisions = [r for r in self.revisions if r.revision != self.revision] + [OrderRevision(
self.purchase_order_id, self.revision, self.net_total, reason, by, datetime.now(timezone.utc),
json.dumps([{"line_no": x.line_no, "quantity": str(x.quantity), "unit_price": str(x.unit_price), "net": str(x.net_amount)}
for x in self.lines]))]
def send(self, to: str, by: str, at: datetime) -> None:
if self.status != "confirmed":
raise RuleBroken("only a confirmed order is sent to the supplier")
self.sends.append(OrderSend(self.purchase_order_id, at, to, by, self.revision))
# ── change and end ────────────────────────────────────────────────────────────────────
def amend(self, changes: dict, new_lines: list, reason: str, by: str, today, received: dict, billed: dict) -> None:
"""Change quantities (`changes`: {line_no: quantity}) and add lines on a confirmed order: a NEW revision, the previous
kept. A line is never reduced below what was received or billed (base unit). The caller decides, from needs_approval, whether
the new total went past what was approved (amend_beyond_approval) or stays within it."""
if self.status != "confirmed":
raise RuleBroken(f"only a confirmed order is amended (this one is {self.status})")
if not (reason or "").strip():
raise RuleBroken("an amendment needs a reason")
for ln, q in changes.items():
x = self.line(ln)
if x is None:
raise RuleBroken(f"the order has no line {ln}")
floor = max(received.get(x.line_no, Fraction(0)), billed.get(x.line_no, Fraction(0)))
if base_of(q, x.unit_numerator, x.unit_denominator) < floor:
raise RuleBroken(f"not_below_received: line {ln} would fall below what was received or billed ({as_dec(floor)} in the "
"base unit): return the goods first (PUR-26) or ask the supplier for a credit")
for ln, q in changes.items():
x = self.line(ln)
x.reprice(self.currency, q)
self.lines.extend(new_lines)
self.revision += 1
self.recompute()
self.snapshot(reason.strip(), by, today)
def move_to_approval(self, today: date, strategy: Strategy) -> None:
self.status, self.submitted_on = "to_approve", today
self.release_levels, self.strategy_valid_from = strategy.levels, strategy.valid_from
self.release_thresholds = json.dumps([str(t) for t in strategy.thresholds])
def cancel(self, reason: str, anything_received_or_billed: bool) -> None:
if self.status in ("closed", "cancelled"):
raise RuleBroken(f"the order is already {self.status}")
if not (reason or "").strip():
raise RuleBroken("cancelling an order needs a reason")
if anything_received_or_billed:
raise RuleBroken("nothing_received: goods were received or billed against this order: close it short instead (PUR-08)")
self.status, self.cancel_reason = "cancelled", reason.strip()
def close_short(self, reason: str) -> None:
if self.status != "confirmed":
raise RuleBroken(f"only a confirmed order is closed short (this one is {self.status})")
if not (reason or "").strip():
raise RuleBroken("closing an order short needs a reason")
self.status, self.close_reason = "closed", reason.strip()
def finish(self) -> None:
if self.status == "confirmed":
self.status = "closed"