dodil/erp-brick-purchasing
DODIL ERP brick: purchasing
src/brick_purchasing/application/orders.py
"""Orders: what we agreed to buy from a named supplier, at a price (PUR-04 to PUR-09, PUR-23, PUR-24, PUR-35, PUR-37, PUR-38, PUR-44, PUR-55).
An order line is ordered in the supplier's own pack (`unit_code`, with the exact conversion copied from product at line time) and priced in its
ORDER PRICE UNIT (`pricing_unit_code`; the conversion is FIXED when the line is priced and never re-read, so a supplier price change after
confirmation re-prices nothing). The readiness panel lists every gate as {gate, met, reason, fix}. Received, billed and outstanding are
DERIVED from receipt and bill rows."""
from datetime import date
from decimal import Decimal
from fractions import Fraction
from typing import Optional
from erp_platform import auth, work
from erp_platform.bricks import Refused
from erp_platform.types import BigId, IsoDate, Limit, text
from brick_purchasing.application import base, rules
from brick_purchasing.application.base import (
ORD_APPROVE,
ORD_CONFIRM,
ORD_EXPECTED,
ORD_READ,
ORD_SEND,
ORD_WRITE,
PURCHASING,
_decide,
_ids,
_same_version,
_stamp,
_sub,
_today,
_who,
stamp_number,
unit,
view,
)
from brick_purchasing.application.inputs import (
ChangeIn,
Code,
Currency,
Offset,
OrderLineIn,
Reason,
Ref,
Version,
fit_money,
)
from brick_purchasing.application.master import price_in_force, profile_on
from brick_purchasing.domain import approval
from brick_purchasing.domain.errors import RuleBroken
from brick_purchasing.domain.order import DeliveryCost, OrderComponent, OrderLine, PurchaseOrder
from brick_purchasing.domain.readiness import no, ok, require, views
from brick_purchasing.domain.values import ZERO, as_dec, fr
from brick_purchasing.infrastructure.seams import Missing, ports
repository = lambda: base.repository() # noqa: E731
# Hooks the agreements module fills (release orders claim a contract; a call-off opens from a JIT release): no-ops until it is loaded.
AGREEMENT_GATES = lambda repo, o, today: [] # noqa: E731
AGREEMENT_CLAIM = lambda repo, o, today: None # noqa: E731
def _load(repo, caller: dict, purchase_order_id: str) -> tuple:
found = repo.orders.get(purchase_order_id)
if found is None or not auth.has_permission(caller, ORD_READ):
raise Refused(404, f"no purchase order {purchase_order_id}")
return found
def _moment() -> str:
"""The moment an order was asked for, with a short random part so two asked in the same instant name two orders."""
import uuid
return work.now().strftime('%Y%m%dT%H%M%S%f') + '-' + uuid.uuid4().hex[:6]
def _supplier_known(caller: dict, bp_id: int) -> None:
if not ports.installed("bp"):
return
try:
ports.bp.get(caller, bp_id)
except Missing as m:
if m.status == 404:
raise Refused(422, f"supplier_known: the supplier {bp_id} is not in the partner register; fix: add it (promote_supplier, role supplier)") from None
raise base._refuse(m, f"supplier {bp_id}") from None
# ═══ pricing a line ═══════════════════════════════════════════════════════════════════════════════
def _convert(caller: dict, product: Optional[dict], product_id: str, unit_code: str, what: str) -> tuple:
"""(numerator, denominator) of `unit_code` to the product's base unit, exact: from the product's units, else product's convert route. A unit
the product cannot convert exactly is refused NAMING the unit."""
if product is None:
return 1, 1
got = base._unit_conv(caller, product, unit_code)
if got is not None:
return got
try:
c = ports.product.convert(caller, product_id, "1", unit_code, product["base_unit_code"])
except Missing as m:
raise Refused(422, f"{what} {unit_code!r} of {product_id} has no exact conversion to {product.get('base_unit_code')}: {m.reason}; "
"fix: Product: add the unit with its conversion") from None
n, _, d = str(c["exact"]).partition("/")
return int(n), int(d or 1)
def _rounding(caller: dict, product_id: str, x_quantity: Decimal, unit_code: str, sp: Optional[dict], today: date) -> tuple:
"""(met, reason, fix) for the purchase rounding: the supplier price row's own multiple where it sets one, else the product's rule."""
if sp and sp.get("order_multiple"):
m = Decimal(str(sp["order_multiple"]))
ok_ = fr(x_quantity) % fr(m) == 0
if ok_:
return True, "", ""
down = (fr(x_quantity) // fr(m)) * m if hasattr(m, "__mul__") else None
lo, hi = as_dec(Fraction(int(fr(x_quantity) // fr(m))) * fr(m)), as_dec(Fraction(int(fr(x_quantity) // fr(m)) + 1) * fr(m))
return (False, f"{product_id} is bought from this supplier in multiples of {m.normalize():f} {unit_code}: {x_quantity.normalize():f} is not one",
f"order {lo.normalize():f} or {hi.normalize():f}, or a second person accepts {x_quantity.normalize():f} with a reason")
if not ports.installed("product") or not product_id:
return True, "", ""
try:
# shared.yaml quantity: a string of at most 4 places; a stored quantity comes back with 6 (100.000000), which product refuses
r = ports.product.rounding(caller, product_id, f"{Decimal(x_quantity).normalize():f}", unit_code, today)
except Missing as m:
if m.status in (404, 422):
return True, "", ""
raise base._refuse(m, f"product {product_id}") from None
if r.get("met", True):
return True, "", ""
down, up = r.get("next_down"), r.get("next_up")
options = " or ".join(str(v) for v in (down, up) if v not in (None, "", "0", 0))
return (False, r.get("reason") or f"{product_id}: {x_quantity} {unit_code} is off its purchase rounding",
(f"order {options}, or " if options else "") + f"a second person accepts {x_quantity.normalize():f} with a reason")
def price_line(repo, caller: dict, o: PurchaseOrder, x: OrderLineIn, line_no: int, today: date) -> OrderLine:
"""One order line, priced. The unit and price default from the supplier's price row in force today (the product master is untouched); the
order price unit's conversion is the supplier row's own where it states one, else product's; quantity_base, quantity_priced and the net are
exact fractions rounded ONCE. The source list is checked: a blocked or unlisted source is refused with the gate source_allowed."""
pid = x.product_id or ""
if not pid and not (x.description or "").strip():
raise Refused(422, f"line {line_no}: name a product or describe what is bought")
product = None
if pid and ports.installed("product"):
try:
product = ports.product.get(caller, pid, today)
except Missing as m:
raise (Refused(422, f"line {line_no}: there is no product {pid}") if m.status == 404 else base._refuse(m, f"product {pid}")) from None
g = rules.source_gate(repo, pid, o.bp_id, today)
if not g.met:
raise Refused(409, f"{g.gate}: {g.reason} (fix: {g.fix})")
item = x.item_category or "standard"
unit_code = x.unit_code or (base.purchase_role_unit(product) if product else "")
probe = None
if pid and not unit_code:
rows = [r for r in repo.supplier_prices.all(bp_id=o.bp_id, product_id=pid) if r["valid_from"] <= today]
unit_code = sorted(r["unit_code"] for r in rows)[0] if rows else ""
if not unit_code:
raise Refused(422, f"line {line_no}: name the unit it is ordered in (the product has no purchase unit)")
sp = price_in_force(repo, o.bp_id, pid, unit_code, x.quantity, today) if pid and item != "consignment" else None
if item == "consignment":
probe = price_in_force(repo, o.bp_id, pid, unit_code, x.quantity, today, consignment=True)
n, d = _convert(caller, product, pid, unit_code, "the unit")
base_unit = product["base_unit_code"] if product else unit_code
# price and order price unit
if item == "consignment":
price, ps, pq, source = ZERO, x.pricing_unit_code or (probe or {}).get("pricing_unit_code") or unit_code, 1, "consignment"
elif x.unit_price is not None:
price, ps, pq, source = x.unit_price, x.pricing_unit_code or (sp or {}).get("pricing_unit_code") or unit_code, \
x.pricing_quantity or (sp or {}).get("pricing_quantity") or 1, "manual"
elif sp:
price, ps, pq, source = Decimal(str(sp["unit_price"])), sp.get("pricing_unit_code") or unit_code, sp.get("pricing_quantity") or 1, "supplier_price"
else:
raise Refused(422, f"line {line_no}: there is no price for {pid or x.description} in {unit_code} from supplier {o.bp_id} in force today "
f"({today}); fix: record the supplier price (record_supplier_price) or give unit_price")
if x.pricing_unit_code and sp and source == "supplier_price":
ps = x.pricing_unit_code
if ps == unit_code:
pn, pd, psrc = n, d, "unit"
elif sp and source == "supplier_price" and sp.get("pricing_numerator") and ps == (sp.get("pricing_unit_code") or unit_code):
pn, pd, psrc = int(sp["pricing_numerator"]), int(sp["pricing_denominator"]), "supplier_price"
else:
pn, pd = _convert(caller, product, pid, ps, "the order price unit")
psrc = "product"
if product is None and ps != unit_code:
raise Refused(422, f"line {line_no}: the order price unit {ps!r} cannot be converted without a product")
met, why, fix = _rounding(caller, pid, x.quantity, unit_code, sp, today) if (pid and item != "consignment" and (x.item_category or "standard") != "subcontract") \
else (True, "", "")
desc = (x.description or (product or {}).get("description") or (product or {}).get("name") or pid).strip()
kind = x.kind or ("service" if (product or {}).get("kind") == "service" else ("service" if not pid else "goods"))
asg = x.account_assignment or "stock"
if asg == "asset" and not (x.asset_id or x.asset_class_code):
raise Refused(422, f"line {line_no}: account assignment asset needs an asset_id, or an asset_class_code to create one at hand-over")
if asg == "cost_centre" and not x.cost_centre:
raise Refused(422, f"line {line_no}: account assignment cost_centre needs a cost_centre")
if asg == "project" and not x.project_id:
raise Refused(422, f"line {line_no}: account assignment project needs a project_id")
line = OrderLine(o.purchase_order_id, line_no, pid, desc, x.quantity, unit_code, n, d, ZERO, base_unit, ps, pn, pd, psrc, int(pq), price, ZERO, ZERO,
source, (product or {}).get("tax_category") or "", x.expense_category or "general", kind, base._day(x.expected_on, "expected_on") or o.expected_on,
asg, x.cost_centre or "", x.asset_id or "", x.asset_class_code or "", x.project_id or "", x.company_code or o.company_code,
item, "", None, bool(x.ers), (Decimal(str(sp["order_multiple"])) if sp and sp.get("order_multiple") else None), met, why, fix,
x.quota_reason or "", (sp or {}).get("supplier_product_code") or "")
line.reprice(o.currency)
fit_money(line.net_amount, f"line {line_no}'s net")
return line
def _apply_lines(repo, caller: dict, o: PurchaseOrder, lines_in: list, today: date) -> None:
priced, comps, costs = [], [], []
for n, x in enumerate(lines_in, 1):
priced.append(price_line(repo, caller, o, x, n, today))
if x.item_category == "subcontract" and x.components:
comps += [OrderComponent(o.purchase_order_id, n, k, c.product_id, c.quantity_per, c.unit_code) for k, c in enumerate(x.components, 1)]
for k, c in enumerate(x.delivery_costs or [], 1):
costs.append(DeliveryCost(o.purchase_order_id, c.line_no or n, c.kind, k, c.amount, c.percent, c.payee_bp_id, o.currency, c.tax_code or ""))
o.set_lines(priced)
o.components = comps
o.delivery_costs = costs
# ═══ gates and the view ════════════════════════════════════════════════════════════════════════════
def asset_gates(repo, caller: dict, o: PurchaseOrder) -> list:
out = []
for x in o.lines:
if x.account_assignment != "asset":
continue
if not ports.installed("fa"):
out.append(no("asset_assignable", "fixed-assets is not part of this installation: account assignment asset is not offered", "expense the line (cost_centre)"))
continue
if not x.asset_id:
try:
classes = ports.fa.classes(_today(repo)).get("classes") or []
except Missing as m:
out.append(no("asset_assignable", f"fixed-assets: {m.reason}", "try again"))
continue
if not any(c["class_code"] == x.asset_class_code for c in classes):
out.append(no("asset_assignable", f"line {x.line_no}: there is no asset class {x.asset_class_code}", "pick another class, or expense the line (cost_centre)"))
continue
try:
a = ports.fa.asset(x.asset_id)
except Missing as m:
out.append(no("asset_assignable", f"line {x.line_no}: asset {x.asset_id}: {m.reason}", "pick another asset or class, or expense the line (cost_centre)"))
continue
bad = []
if a.get("status") == "retired":
bad.append("is retired")
if x.company_code and a.get("company_code") and a["company_code"] != x.company_code:
bad.append(f"belongs to company {a['company_code']}")
thr = a.get("capitalisation_threshold")
if thr not in (None, "") and x.net_amount < Decimal(str(thr)) and a.get("class_code") != "LVA":
bad.append(f"is below the capitalisation threshold {thr}")
if bad:
out.append(no("asset_assignable", f"line {x.line_no}: asset {x.asset_id} " + ", ".join(bad), "pick another asset or class, or expense the line (cost_centre)"))
return out or ([ok("asset_assignable")] if any(x.account_assignment == "asset" for x in o.lines) else [])
def confirm_gates(repo, caller: dict, o: PurchaseOrder, today: date) -> list:
"""Every gate of confirm_purchase_order: authority, units, pricing units, purchase rounding (with its overrides), the source list, the
contract (a release order fits what is left), asset assignment, subcontract components, the number series."""
s = base.Settings(repo, today)
gates = o.core_gates(s["order_approval_threshold"])
seen = set()
for x in o.lines:
g = rules.source_gate(repo, x.product_id, o.bp_id, today)
if not g.met and ("source_allowed", x.line_no) not in seen:
gates.append(g)
seen.add(("source_allowed", x.line_no))
if not any(g.gate == "source_allowed" for g in gates):
gates.append(ok("source_allowed"))
gates += AGREEMENT_GATES(repo, o, today)
gates += asset_gates(repo, caller, o)
fmt = repo.settings.value_on("order_number_format", today)
gates.append(ok("number_series") if fmt and "{seq" in fmt else no("number_series", "no order number series for this year", "Settings: number format (owner)"))
return gates
def order_view(repo, caller: dict, o: PurchaseOrder, version: int, *, gates: bool = True) -> dict:
out = view(o, drop=("components", "delivery_costs", "releases", "sends", "revisions", "overrides"))
recv = rules.received_by_line(repo, o.purchase_order_id)
billed = rules.billed_by_line(repo, o.purchase_order_id)
for ln in out["lines"]:
x = o.line(ln["line_no"])
ln.update(rules.line_status(x, recv.get(x.line_no, Fraction(0)), billed.get(x.line_no, Fraction(0))))
out["total"] = o.net_total
out["version"] = version
out["waiting"] = o.waiting()
out["approved_levels"] = len([r for r in o.releases if r.revision == o.revision])
out["releases"] = [view(r) for r in o.releases]
out["revisions"] = [view(r) for r in o.revisions]
out["sends"] = [view(r) for r in o.sends]
out["overrides"] = [view(r) for r in o.overrides]
out["components"] = [view(r) for r in o.components]
out["delivery_costs"] = [view(r) for r in o.delivery_costs]
if gates and o.status in ("draft", "approved"):
out["readiness"] = views(confirm_gates(repo, caller, o, _today(repo)))
return out
# ═══ drafting ════════════════════════════════════════════════════════════════════════════════════
def _new_order(repo, caller, pid: str, bp_id: int, currency: str, **kw) -> PurchaseOrder:
return PurchaseOrder(pid, bp_id=bp_id, currency=base._currency(currency), buyer_email=_who(caller), buyer_sub=_sub(caller),
terms_id=kw.pop("terms_id", "") or profile_on(repo, bp_id, _today(repo))["terms_id"], **kw)
@PURCHASING.use_case(kind="command", route="POST /purchase-orders", permission=ORD_WRITE)
def draft_order(caller: dict, bp_id: BigId, currency: Currency = "GBP", lines: Optional[list[OrderLineIn]] = None, supplier_ref: Optional[text(100)] = None,
expected_on: Optional[IsoDate] = None, company_code: Optional[text(4)] = None) -> dict:
"""Draft an order directly to a supplier (no request). The supplier must be in the partner register. Lines are priced as price_lines does:
unit, pack and price default from the supplier's own price in force today. Keyed po:<buyer>:<moment>. An order above the owner's threshold
needs an approver before it can be confirmed (PUR-09)."""
_supplier_known(caller, bp_id)
def run():
with repository() as repo, _decide():
today = _today(repo)
pid = f"po:{_who(caller)}:{_moment()}"
o = _new_order(repo, caller, pid, bp_id, currency, supplier_ref=supplier_ref or "", expected_on=base._day(expected_on, "expected_on"),
company_code=company_code or "", source_kind="direct", source_id=pid)
if lines:
_apply_lines(repo, caller, o, lines, today)
return order_view(repo, caller, o, repo.orders.save(o, 0))
return unit(None, run)
@PURCHASING.use_case(kind="command", route="POST /purchase-requests/{purchase_request_id}/raise-order", permission=ORD_WRITE, idempotent=True)
def raise_order(caller: dict, purchase_request_id: Ref, bp_id: Optional[BigId] = None, lines: Optional[list[OrderLineIn]] = None,
currency: Optional[Currency] = None, supplier_ref: Optional[text(100)] = None, quota_reason: Optional[Reason] = None) -> dict:
"""Raise the draft order from an APPROVED request and a chosen supplier. Without lines, the request's lines are ordered (priced from the
supplier's price). The approval travels with the order; a second raise from the same request returns the same order. Raising from an unapproved
request is refused. Without a supplier, the quota arrangement proposes one (PUR-45) — choosing another needs a reason, kept on the line."""
def run():
with repository() as repo, _decide():
got = repo.requests.get(purchase_request_id)
if got is None or not auth.has_permission(caller, base.REQ_READ):
raise Refused(404, f"no purchase request {purchase_request_id}")
r, rv = got
oid = f"req:{purchase_request_id}"
have = repo.orders.get(oid)
if have is not None:
return order_view(repo, caller, have[0], have[1])
if r.status != "approved":
raise RuleBroken(f"a request that is {r.status} cannot be ordered from: raise an order from an APPROVED request")
today = _today(repo)
supplier = bp_id
first = r.lines[0] if r.lines else None
if supplier is None and first and first.product_id:
prop = rules.quota_proposal(repo, first.product_id, first.quantity, today)
supplier = (prop or {}).get("proposed", {}) and prop["proposed"]["bp_id"]
if not supplier:
raise RuleBroken("supplier_known: a supplier is chosen: name the supplier (bp_id)")
_supplier_known(caller, supplier)
o = _new_order(repo, caller, oid, supplier, currency or r.currency, supplier_ref=supplier_ref or "", source_kind="request", source_id=r.purchase_request_id,
request_id=r.purchase_request_id, approved_total=r.amount, approved_by_email=r.approved_by_email, approved_by_sub=r.approved_by_sub,
approved_on=r.approved_on, ship_to_bp_id=r.ship_to_bp_id, ship_to_address=r.ship_to_address, sales_order_id=r.sales_order_id,
so_line_no=r.so_line_no, warehouse_id=r.warehouse_id, location_id=r.location_id, expected_on=r.needed_by)
from_request = lines or [OrderLineIn(product_id=x.product_id or None, description=x.description, quantity=x.quantity, unit_code=x.unit_code or None,
unit_price=x.unit_price if (x.unit_price and not x.product_id) else None,
expense_category=r.category, cost_centre=r.cost_centre or None) for x in r.lines]
_apply_lines(repo, caller, o, from_request, today)
r.mark_ordered(oid)
repo.requests.save(r, rv)
return order_view(repo, caller, o, repo.orders.save(o, 0))
return unit(None, run)
@PURCHASING.use_case(kind="command", route="POST /purchase-orders/{purchase_order_id}/price-lines", permission=ORD_WRITE)
def price_lines(caller: dict, purchase_order_id: Ref, lines: list[OrderLineIn], version: Optional[Version] = None) -> dict:
"""Price the lines of a draft order in the supplier's unit. Each line copies the unit and its exact conversion from the product, takes the
supplier's price, order price unit and its conversion from the supplier price row in force today (or a manual unit_price), and shows the purchase
rounding gate {gate, met, reason, fix}. The conversion is fixed on the line. A blocked or unlisted source is refused (source_allowed)."""
if not lines:
raise Refused(422, "price at least one line")
def run():
with repository() as repo, _decide():
o, v = _load(repo, caller, purchase_order_id)
_same_version("the order", v, version)
_apply_lines(repo, caller, o, lines, _today(repo))
return order_view(repo, caller, o, repo.orders.save(o, v))
return unit(version, run)
@PURCHASING.use_case(kind="command", route="POST /purchase-orders/{purchase_order_id}/accept-rounding", permission=ORD_APPROVE)
def accept_rounding(caller: dict, purchase_order_id: Ref, line_no: int, reason: Reason, version: Optional[Version] = None) -> dict:
"""A second person (a budget holder, never the order's buyer — same email OR same sub) accepts an order line off the pack rule, with a reason:
an override row for this order revision and line. On a confirmed order it applies to the amendment that creates the next revision."""
def run():
with repository() as repo, _decide():
o, v = _load(repo, caller, purchase_order_id)
_same_version("the order", v, version)
rev = o.revision + 1 if o.status == "confirmed" else o.revision
o.accept_rounding(line_no, rev, email=caller.get("email"), sub=caller.get("sub"), reason=reason, today=_today(repo))
return order_view(repo, caller, o, repo.orders.save(o, v))
return unit(version, run)
# ═══ approval ════════════════════════════════════════════════════════════════════════════════════
def _category(o: PurchaseOrder) -> str:
return (o.lines[0].expense_category if o.lines else "") or "general"
def _limit(repo, caller, o: PurchaseOrder, today):
return approval.limit_in_force(repo.limits.all(), _ids(caller), _category(o), o.currency, today)
@PURCHASING.use_case(kind="command", route="POST /purchase-orders/{purchase_order_id}/submit", permission=ORD_WRITE, idempotent=True)
def submit_order(caller: dict, purchase_order_id: Ref, version: Optional[Version] = None) -> dict:
"""Send the order for approval: no approved request covers its total and the total is above the owner's order threshold. It waits for an
approver whose limit in force today covers it, under the release strategy in force today (up to three levels)."""
def run():
with repository() as repo, _decide():
o, v = _load(repo, caller, purchase_order_id)
if o.status == "to_approve":
return order_view(repo, caller, o, v)
_same_version("the order", v, version)
today = _today(repo)
o.submit(today, approval.strategy_for(repo.strategies.all(), _category(o), o.net_total, o.currency, today),
base.Settings(repo, today)["order_approval_threshold"])
return order_view(repo, caller, o, repo.orders.save(o, v))
return unit(version, run)
@PURCHASING.use_case(kind="command", route="POST /purchase-orders/{purchase_order_id}/approve-level", permission=ORD_APPROVE)
def approve_order_level(caller: dict, purchase_order_id: Ref, version: Optional[Version] = None) -> dict:
"""Approve one NON-FINAL release level of an order. Your limit meets the level's threshold; you are not the buyer nor an earlier level's
approver (email AND sub). Each level is a row: order, revision, level, who, when. The order stays to_approve, showing 'level n of m waiting'."""
def run():
with repository() as repo, _decide():
o, v = _load(repo, caller, purchase_order_id)
_same_version("the order", v, version)
today = _today(repo)
o.approve_level(email=caller.get("email"), sub=caller.get("sub"), limit=_limit(repo, caller, o, today), today=today)
return order_view(repo, caller, o, repo.orders.save(o, v))
return unit(version, run)
@PURCHASING.use_case(kind="command", route="POST /purchase-orders/{purchase_order_id}/approve", permission=ORD_APPROVE)
def approve_order(caller: dict, purchase_order_id: Ref, version: Optional[Version] = None) -> dict:
"""Approve the order (the FINAL level): the total is within your limit in force today, and you are not the buyer nor an earlier level's approver
(same email OR same sub). The order can then be confirmed; what you approved is what later amendments are held to."""
def run():
with repository() as repo, _decide():
o, v = _load(repo, caller, purchase_order_id)
_same_version("the order", v, version)
today = _today(repo)
o.approve(email=caller.get("email"), sub=caller.get("sub"), limit=_limit(repo, caller, o, today), today=today)
return order_view(repo, caller, o, repo.orders.save(o, v))
return unit(version, run)
@PURCHASING.use_case(kind="command", route="POST /purchase-orders/{purchase_order_id}/reject", permission=ORD_APPROVE)
def reject_order(caller: dict, purchase_order_id: Ref, reason: Reason, version: Optional[Version] = None) -> dict:
"""Send an order back to the buyer with a reason; the levels approved so far are void."""
def run():
with repository() as repo, _decide():
o, v = _load(repo, caller, purchase_order_id)
_same_version("the order", v, version)
o.reject(reason)
return order_view(repo, caller, o, repo.orders.save(o, v))
return unit(version, run)
# ═══ confirm, send, amend, end ═══════════════════════════════════════════════════════════════════
@PURCHASING.use_case(kind="command", route="POST /purchase-orders/{purchase_order_id}/confirm", permission=ORD_CONFIRM, idempotent=True)
def confirm_purchase_order(caller: dict, purchase_order_id: Ref, version: Optional[Version] = None) -> dict:
"""Confirm the order once every readiness gate is met (get_purchase_order lists them as {gate, met, reason, fix}): within authority, every
line's unit and order price unit convert, every line meets its purchase rounding or carries a second person's override for this revision,
every supplier is allowed by the source list, a release order fits its contract, an asset line is assignable. The order number is stamped ONCE
from the series, the contents and each line's conversion freeze, and product is told every unit and pricing unit was used. A retry returns the
number already stamped."""
def run():
with repository() as repo, _decide():
o, v = _load(repo, caller, purchase_order_id)
if o.status == "confirmed":
return order_view(repo, caller, o, v)
_same_version("the order", v, version)
today = _today(repo)
require(confirm_gates(repo, caller, o, today))
number = o.order_no or stamp_number(repo, "order", "order_number_format", today, o.purchase_order_id)
o.confirm(today, number)
AGREEMENT_CLAIM(repo, o, today)
if ports.installed("product"):
for pid, codes in {x.product_id: {x.unit_code, x.pricing_unit_code} for x in o.lines if x.product_id}.items():
for code in sorted(codes):
try:
ports.product.unit_used(pid, code, f"pur_order:{o.purchase_order_id}")
except Missing as m:
raise Refused(409, f"{m.gate or 'unit_used'}: {pid} {code}: {m.reason}") from None
return order_view(repo, caller, o, repo.orders.save(o, v))
return unit(version, run)
@PURCHASING.use_case(kind="command", route="POST /purchase-orders/{purchase_order_id}/send", permission=ORD_SEND)
def send_order(caller: dict, purchase_order_id: Ref, to: text(254, min_length=3), version: Optional[Version] = None) -> dict:
"""Record that the confirmed order was sent to the supplier, when and to whom. Sending twice records a second send and changes nothing else."""
def run():
with repository() as repo, _decide():
o, v = _load(repo, caller, purchase_order_id)
_same_version("the order", v, version)
o.send(to.strip(), _who(caller), _stamp())
return order_view(repo, caller, o, repo.orders.save(o, v), gates=False)
return unit(version, run)
def _amend(repo, caller, o: PurchaseOrder, v: int, changes, new_lines, reason: str, today: date) -> None:
recv, billed = rules.received_by_line(repo, o.purchase_order_id), rules.billed_by_line(repo, o.purchase_order_id)
extra = [price_line(repo, caller, o, x, max([ln.line_no for ln in o.lines] + [0]) + k, today) for k, x in enumerate(new_lines or [], 1)]
o.amend({c.line_no: c.quantity for c in (changes or [])}, extra, reason, _who(caller), today, recv, billed)
# the purchase rounding of every changed or added line asks again for THIS revision
touched = {c.line_no for c in (changes or [])} | {x.line_no for x in extra}
for ln in touched:
x = o.line(ln)
if x.item_category == "standard" and x.product_id:
sp = price_in_force(repo, o.bp_id, x.product_id, x.unit_code, x.quantity, today)
x.rounding_met, x.rounding_reason, x.rounding_fix = _rounding(caller, x.product_id, x.quantity, x.unit_code, sp, today)
bad = [o.rounding_gate(o.line(ln)) for ln in touched]
require([g for g in bad if not g.met])
@PURCHASING.use_case(kind="command", route="POST /purchase-orders/{purchase_order_id}/amend", permission=ORD_WRITE)
def amend_purchase_order(caller: dict, purchase_order_id: Ref, reason: Reason, changes: Optional[list[ChangeIn]] = None,
add_lines: Optional[list[OrderLineIn]] = None, version: Optional[Version] = None) -> dict:
"""Change quantities (or add lines) on a confirmed order, with a reason: a NEW revision, the previous one kept. A line keeps its fixed
conversion and price. The new total must still be within what was approved (else use amend_beyond_approval); a line is never reduced below
what was received or billed; a changed line that is off its purchase rounding asks for a second person's override again."""
def run():
with repository() as repo, _decide():
o, v = _load(repo, caller, purchase_order_id)
_same_version("the order", v, version)
_amend(repo, caller, o, v, changes, add_lines, reason, _today(repo))
s = base.Settings(repo, _today(repo))
if o.needs_approval(s["order_approval_threshold"]):
raise RuleBroken("the new total goes beyond what was approved: use amend_beyond_approval to send the change for approval")
return order_view(repo, caller, o, repo.orders.save(o, v), gates=False)
return unit(version, run)
@PURCHASING.use_case(kind="command", route="POST /purchase-orders/{purchase_order_id}/amend-beyond-approval", permission=ORD_WRITE)
def amend_beyond_approval(caller: dict, purchase_order_id: Ref, reason: Reason, changes: Optional[list[ChangeIn]] = None,
add_lines: Optional[list[OrderLineIn]] = None, version: Optional[Version] = None) -> dict:
"""Amend a confirmed order so that its total goes beyond what was approved: a new revision that goes back to approval (to_approve) under the
release strategy in force today. Once approved it is confirmed again; it keeps its order number."""
def run():
with repository() as repo, _decide():
o, v = _load(repo, caller, purchase_order_id)
_same_version("the order", v, version)
today = _today(repo)
_amend(repo, caller, o, v, changes, add_lines, reason, today)
s = base.Settings(repo, today)
if not o.needs_approval(s["order_approval_threshold"]):
raise RuleBroken("the new total is still within what was approved: use amend_purchase_order")
o.move_to_approval(today, approval.strategy_for(repo.strategies.all(), _category(o), o.net_total, o.currency, today))
return order_view(repo, caller, o, repo.orders.save(o, v))
return unit(version, run)
@PURCHASING.use_case(kind="command", route="POST /purchase-orders/{purchase_order_id}/close-short", permission=ORD_WRITE, idempotent=True)
def close_short(caller: dict, purchase_order_id: Ref, reason: Reason, version: Optional[Version] = None) -> dict:
"""Close an order that will never be completed: what was not received stops being outstanding and stops being a commitment. The reason is recorded."""
def run():
with repository() as repo, _decide():
o, v = _load(repo, caller, purchase_order_id)
if o.status == "closed":
return order_view(repo, caller, o, v, gates=False)
_same_version("the order", v, version)
o.close_short(reason)
return order_view(repo, caller, o, repo.orders.save(o, v), gates=False)
return unit(version, run)
@PURCHASING.use_case(kind="command", route="POST /purchase-orders/{purchase_order_id}/cancel", permission=ORD_WRITE, idempotent=True, destructive=True)
def cancel_purchase_order(caller: dict, purchase_order_id: Ref, reason: Reason, version: Optional[Version] = None) -> dict:
"""Cancel an order before anything arrives: it keeps its number (if confirmed), the reason is recorded and no bill can be matched to it. An
order with receipts or bills against it cannot be cancelled, only closed short."""
def run():
with repository() as repo, _decide():
o, v = _load(repo, caller, purchase_order_id)
if o.status == "cancelled":
return order_view(repo, caller, o, v, gates=False)
_same_version("the order", v, version)
touched = bool(repo.receipts.lines_of_order(o.purchase_order_id)) or bool(repo.bills.of_order(o.purchase_order_id))
o.cancel(reason, touched)
saved = repo.orders.save(o, v)
if o.request_id: # v8 (PUR-67) cancel_with_order: the request ends with its order, the order's reason copied
got = repo.requests.get(o.request_id)
if got and got[0].status == "ordered":
got[0].cancel(o.cancel_reason, email=_who(caller), sub=_sub(caller), today=_today(repo), with_order=True)
repo.requests.save(got[0], got[1])
return order_view(repo, caller, o, saved, gates=False)
return unit(version, run)
# ═══ reading ═══════════════════════════════════════════════════════════════════════════════════════
@PURCHASING.use_case(kind="query", route="GET /purchase-orders/{purchase_order_id}", permission=ORD_READ, etag="version")
def get_purchase_order(caller: dict, purchase_order_id: Ref) -> dict:
"""One order with its lines (received, billed and outstanding per line, DERIVED), its release levels, revisions, sends and the readiness gates
of the next transition as {gate, met, reason, fix}."""
with repository() as repo:
o, v = _load(repo, caller, purchase_order_id)
return order_view(repo, caller, o, v)
@PURCHASING.use_case(kind="query", route="GET /purchase-orders", permission=ORD_READ)
def list_purchase_orders(caller: dict, bp_id: Optional[BigId] = None, status: Optional[Code] = None, limit: Limit = 50, offset: Offset = 0) -> dict:
"""The orders (heads: supplier, number, status, total), newest first."""
with repository() as repo:
rows, total = repo.orders.page(bp_id=bp_id, status=status, limit=limit, offset=offset)
return {"total": total, "orders": [{k: getattr(o, k) for k in ("purchase_order_id", "order_no", "bp_id", "currency", "status", "net_total",
"confirmed_on", "expected_on", "source_kind", "revision")} for o in rows]}
@PURCHASING.use_case(kind="query", route="GET /purchase-orders/{purchase_order_id}/expected", permission=ORD_EXPECTED, tool=False)
def get_expected(caller: dict, purchase_order_id: Ref) -> dict:
"""What is still to arrive on an order (inventory's incoming screen and the receiver): per line the ordered, received and outstanding quantity."""
with repository() as repo:
got = repo.orders.get(purchase_order_id)
if got is None:
raise Refused(404, f"no purchase order {purchase_order_id}")
o = got[0]
recv = rules.received_by_line(repo, purchase_order_id)
lines = []
for x in o.lines:
st = rules.line_status(x, recv.get(x.line_no, Fraction(0)), Fraction(0))
lines.append({"line_no": x.line_no, "product_id": x.product_id, "unit_code": x.unit_code, "ordered": x.quantity, "received": st["received"],
"outstanding": st["outstanding"], "quantity_base_outstanding": st["outstanding_base"], "expected_on": x.expected_on})
return {"purchase_order_id": o.purchase_order_id, "order_no": o.order_no, "bp_id": o.bp_id, "lines": lines}
@PURCHASING.use_case(kind="query", route="GET /expected-receipts", permission=ORD_EXPECTED, tool=False)
def list_expected_receipts(caller: dict, product_id: Optional[Code] = None, warehouse_id: Optional[Code] = None) -> list:
"""What is still to arrive per product and warehouse, from confirmed order lines less receipts; inventory's ATP and replenishment read it
and keep no copy. Subcontract and consignment lines name their special stock."""
with repository() as repo:
out = []
recvs = {}
for o, x in repo.orders.open_lines(product_id):
if warehouse_id and o.warehouse_id and o.warehouse_id != warehouse_id:
continue
if o.purchase_order_id not in recvs:
recvs[o.purchase_order_id] = rules.received_by_line(repo, o.purchase_order_id)
left = max(x.exact_base - recvs[o.purchase_order_id].get(x.line_no, Fraction(0)), Fraction(0))
if left > 0 and x.kind == "goods" and x.account_assignment == "stock":
out.append({"purchase_order_id": o.purchase_order_id, "line_no": x.line_no, "expected_on": x.expected_on,
"quantity_base_outstanding": as_dec(left), "special_stock": {"consignment": "K", "subcontract": "O"}.get(x.item_category, "")})
return out