dodil/erp-brick-purchasing
DODIL ERP brick: purchasing
src/brick_purchasing/application/agreements.py
"""Outline agreements: quantity and value contracts (PUR-41, PUR-42).
A contract is approved ONCE under the release strategy; that approval is the spend approval for every release order within its target and validity,
so a release order is drafted already covered and confirms without asking again. What has been released is never stored as a counter: it is derived
from claim rows (pur_agreement_claim, one per agreement line and order line), written in the same transaction as the confirmation and serialised on the
agreement by a claim sequence, so two buyers confirming at the same moment cannot both pass the last of the target."""
import json
from datetime import date
from decimal import Decimal
from fractions import Fraction
from typing import Literal, Optional
from erp_platform import auth
from erp_platform.bricks import Refused
from erp_platform.types import BigId, IsoDate, text
from brick_purchasing.application import base, orders
from brick_purchasing.application.base import (
AGR_APPROVE,
AGR_WRITE,
ORD_WRITE,
PURCHASING,
_decide,
_ids,
_same_version,
_sub,
_today,
_who,
unit,
view,
)
from brick_purchasing.application.inputs import Code, Money, Price, Qty, Reason, Ref, Version, Whole, _In
from brick_purchasing.application.master import check_start # noqa: F401
from brick_purchasing.domain import approval, rows
from brick_purchasing.domain.approval import identities
from brick_purchasing.domain.errors import RuleBroken
from brick_purchasing.domain.readiness import no, ok, require
from brick_purchasing.domain.values import ZERO, as_dec, fr, money_round
from brick_purchasing.infrastructure.seams import ports
repository = lambda: base.repository() # noqa: E731
class AgreementLineIn(_In):
product_id: Code
unit_code: Optional[Code] = None
target_quantity: Optional[Qty] = None
unit_price: Price
pricing_quantity: Optional[Whole] = None
def _line_unit(caller: dict, product_id: str, unit_code: Optional[str], today: date) -> tuple:
product = ports.product.get(caller, product_id, today) if ports.installed("product") else None
code = unit_code or (base.purchase_role_unit(product) if product else "")
if not code:
raise Refused(422, f"{product_id}: name the unit the agreement is priced in")
n, d = orders._convert(caller, product, product_id, code, "the unit")
return code, n, d, (product or {}).get("base_unit_code") or code
def target_of(a: rows.PurchaseAgreement) -> Decimal:
"""What approval is for: the value of a value contract, else the priced quantities of a quantity contract."""
if a.kind == "value_contract":
return a.target_value
return money_round(sum((fr(x.target_quantity) * fr(x.unit_price) / int(x.pricing_quantity or 1) for x in a.lines), Fraction(0)), a.currency)
def claimed(repo, agreement_id: str) -> dict:
"""{agreement line_no: (quantity_base, value)} released so far, derived from the claim rows."""
out = {}
for c in repo.agreement_claims.all(agreement_id=agreement_id):
q, v = out.get(c["line_no"], (Fraction(0), Fraction(0)))
out[c["line_no"]] = (q + fr(c["quantity_base"] or 0), v + fr(c["value"] or 0))
return out
def agreement_view(repo, a: rows.PurchaseAgreement, version: int) -> dict:
out = view(a, drop=("releases", "schedule", "releases_sent"))
c = claimed(repo, a.agreement_id)
out["version"], out["target"] = version, target_of(a)
out["released_value"] = as_dec(sum((v for _, v in c.values()), Fraction(0)), 2)
out["remaining_value"] = as_dec(fr(target_of(a)) - fr(out["released_value"]), 2) if a.kind == "value_contract" else None
for ln in out["lines"]:
q = c.get(ln["line_no"], (Fraction(0), Fraction(0)))[0]
target = fr(ln["target_quantity"]) * Fraction(ln["unit_numerator"], ln["unit_denominator"])
ln["released_base"], ln["remaining_base"] = as_dec(q, 6), (as_dec(max(target - q, 0), 6) if a.kind == "quantity_contract" else None)
out["approved_levels"] = len([r for r in a.releases if r.revision == a.revision])
out["releases"] = [view(r) for r in a.releases]
return out
def _load(repo, caller: dict, aid: str) -> tuple:
got = repo.agreements.get(aid)
if got is None or not auth.has_permission(caller, AGR_WRITE) and not auth.has_permission(caller, AGR_APPROVE):
raise Refused(404, f"no agreement {aid}")
return got
def _lines(caller: dict, aid: str, lines: list, today: date) -> list:
out = []
for n, x in enumerate(lines, 1):
code, un, ud, base_unit = _line_unit(caller, x.product_id, x.unit_code, today)
out.append(rows.AgreementLine(aid, n, x.product_id, "", code, un, ud, x.target_quantity or ZERO, x.unit_price, code, un, ud, int(x.pricing_quantity or 1), base_unit))
return out
@PURCHASING.use_case(kind="command", route="POST /agreements", permission=AGR_WRITE, idempotent=True)
def draft_agreement(caller: dict, bp_id: BigId, kind: Literal["quantity_contract", "value_contract", "scheduling_agreement"], supplier_ref: Code, valid_from: IsoDate, valid_to: IsoDate,
lines: list[AgreementLineIn], target_value: Optional[Money] = None, currency: Optional[text(3, min_length=3)] = None) -> dict:
"""Draft a contract with a supplier for a quantity or a value over a period, with the agreed prices. Keyed ag:<bp_id>:<kind>:<valid_from>:<the
supplier's reference>; drafting it again returns the same one. A value contract needs its target value; a quantity contract a target quantity on
every line. The agreement number comes from the series when it is approved."""
orders._supplier_known(caller, bp_id)
def run():
with repository() as repo, _decide():
today = _today(repo)
start, end = base._day(valid_from, "valid_from"), base._day(valid_to, "valid_to")
if end < start:
raise Refused(422, "valid_to is before valid_from")
aid = f"ag:{int(bp_id)}:{kind}:{start}:{supplier_ref}"
have = repo.agreements.get(aid)
if have is not None:
return agreement_view(repo, *have)
if kind == "value_contract" and not target_value:
raise Refused(422, "a value contract needs a target_value")
if kind != "value_contract" and any(not x.target_quantity for x in lines):
raise Refused(422, "a quantity contract or scheduling agreement needs a target_quantity on every line")
a = rows.PurchaseAgreement(aid, int(bp_id), kind, supplier_ref, valid_from=start, valid_to=end, target_value=target_value or ZERO,
currency=base._currency(currency or "GBP"), buyer_email=_who(caller), buyer_sub=_sub(caller))
a.lines = _lines(caller, aid, lines, today)
return agreement_view(repo, a, repo.agreements.save(a, 0))
return unit(None, run)
@PURCHASING.use_case(kind="command", route="POST /agreements/{agreement_id}/submit", permission=AGR_WRITE, idempotent=True)
def submit_agreement(caller: dict, agreement_id: Ref, version: Optional[Version] = None) -> dict:
"""Send the contract (or a revision that raised its target or a price) for approval under the release strategy in force today."""
def run():
with repository() as repo, _decide():
a, v = _load(repo, caller, agreement_id)
_same_version("the agreement", v, version)
if a.status == "to_approve":
return agreement_view(repo, a, v)
if a.status not in ("draft", "active") or (a.status == "active" and not a.pending_json):
raise RuleBroken(f"an agreement that is {a.status} has nothing to approve")
today = _today(repo)
s = approval.strategy_for(repo.strategies.all(), "general", target_of(a), a.currency, today)
a.release_levels, a.strategy_valid_from, a.submitted_on = s.levels, s.valid_from, today
a.release_thresholds = json.dumps([str(t) for t in s.thresholds])
if a.status == "draft":
a.status = "to_approve"
return agreement_view(repo, a, repo.agreements.save(a, v))
return unit(version, run)
def _step(repo, caller: dict, a: rows.PurchaseAgreement, final: bool) -> None:
if a.status not in ("to_approve", "active") or (a.status == "active" and not a.pending_json):
raise RuleBroken(f"an agreement that is {a.status} is not waiting for approval")
if not a.submitted_on:
raise RuleBroken("send the agreement for approval first (submit_agreement)")
today = _today(repo)
limit = approval.limit_in_force(repo.limits.all(), _ids(caller), "general", a.currency, today)
rev = a.revision + (1 if a.pending_json else 0)
approval.step(a, "agreement", a.agreement_id, rev, owner_ids=identities(a.buyer_email, a.buyer_sub), email=caller.get("email"), sub=caller.get("sub"),
limit=limit, needed=target_of(a) if not a.pending_json else Decimal(json.loads(a.pending_json)["target"]), today=today, noun="agreement", final=final)
@PURCHASING.use_case(kind="command", route="POST /agreements/{agreement_id}/approve-level", permission=AGR_APPROVE)
def approve_agreement_level(caller: dict, agreement_id: Ref, version: Optional[Version] = None) -> dict:
"""Approve one non-final release level of the agreement (not the buyer, not an earlier level's approver: email AND sub)."""
def run():
with repository() as repo, _decide():
a, v = _load(repo, caller, agreement_id)
_same_version("the agreement", v, version)
_step(repo, caller, a, False)
return agreement_view(repo, a, repo.agreements.save(a, v))
return unit(version, run)
@PURCHASING.use_case(kind="command", route="POST /agreements/{agreement_id}/approve", permission=AGR_APPROVE)
def approve_agreement(caller: dict, agreement_id: Ref, version: Optional[Version] = None) -> dict:
"""Approve the agreement (the FINAL level): your limit covers its target, and you are not the buyer (email OR sub). The number is stamped from
the series. This approval is the spend approval for every release order within the target and the validity. A pending revision becomes the
approved one; until then releases continue under the approved revision."""
def run():
with repository() as repo, _decide():
a, v = _load(repo, caller, agreement_id)
_same_version("the agreement", v, version)
_step(repo, caller, a, True)
today = _today(repo)
if a.pending_json:
p = json.loads(a.pending_json)
a.revision += 1
a.target_value = Decimal(p["target_value"])
prices = {int(k): Decimal(x) for k, x in p["prices"].items()}
for ln in a.lines:
ln.unit_price = prices.get(ln.line_no, ln.unit_price)
a.pending_json = ""
a.agreement_no = a.agreement_no or base.stamp_number(repo, "agreement", "agreement_number_format", today, a.agreement_id)
a.status, a.approved_on = "active", today
a.approved_by_email, a.approved_by_sub = base._email(caller), _sub(caller)
return agreement_view(repo, a, repo.agreements.save(a, v))
return unit(version, run)
@PURCHASING.use_case(kind="command", route="POST /agreements/{agreement_id}/revise", permission=AGR_WRITE)
def revise_agreement(caller: dict, agreement_id: Ref, target_value: Optional[Money] = None, prices: Optional[dict[str, Price]] = None,
version: Optional[Version] = None) -> dict:
"""Propose a higher target or other prices (by line number). It goes back to approval; releases continue under the approved revision until
it is approved. A revision that lowers nothing and raises nothing is refused."""
def run():
with repository() as repo, _decide():
a, v = _load(repo, caller, agreement_id)
_same_version("the agreement", v, version)
if a.status != "active":
raise RuleBroken(f"only an approved agreement is revised: this one is {a.status}")
new_target = target_value if target_value is not None else a.target_value
new_prices = {str(k): str(x) for k, x in (prices or {}).items()}
if a.kind == "value_contract" and new_target < a.target_value:
raise Refused(409, "the target cannot go below the approved target in a revision")
a.pending_json = json.dumps({"target_value": str(new_target), "prices": new_prices, "target": str(new_target if a.kind == "value_contract" else target_of(a))})
a.submitted_on = None
a.releases = [r for r in a.releases if r.revision != a.revision + 1]
return agreement_view(repo, a, repo.agreements.save(a, v))
return unit(version, run)
@PURCHASING.use_case(kind="query", route="GET /agreements/{agreement_id}", permission=AGR_WRITE)
def get_agreement(caller: dict, agreement_id: Ref) -> dict:
"""The agreement with what has been released and what is left, both derived from the claim rows."""
with repository() as repo:
return agreement_view(repo, *_load(repo, caller, agreement_id))
class ReleaseLineIn(_In):
line_no: int
quantity: Qty
@PURCHASING.use_case(kind="command", route="POST /agreements/{agreement_id}/release-orders", permission=ORD_WRITE)
def draft_release_order(caller: dict, agreement_id: Ref, lines: list[ReleaseLineIn], expected_on: Optional[IsoDate] = None) -> dict:
"""Draft an order against an approved agreement: the lines are priced at the agreement's prices (they take precedence over supplier prices) and
the order is already covered by the agreement's approval. It must fit what is left (checked at confirm: the gate within_agreement)."""
def run():
with repository() as repo, _decide():
a, _ = _load(repo, caller, agreement_id)
today = _today(repo)
if a.status != "active":
raise RuleBroken(f"an agreement that is {a.status} takes no release orders: it must be approved")
if today > a.valid_to or today < a.valid_from:
raise Refused(409, f"within_agreement: the agreement runs {a.valid_from} to {a.valid_to}; today is {today}")
pid = f"ao:{agreement_id}:{orders._moment()}"
o = orders._new_order(repo, caller, pid, a.bp_id, a.currency, source_kind="agreement", source_id=agreement_id, agreement_id=agreement_id,
approved_by_email=a.approved_by_email, approved_by_sub=a.approved_by_sub, approved_on=a.approved_on,
expected_on=base._day(expected_on, "expected_on"))
lines_in = []
for x in lines:
al = next((ln for ln in a.lines if ln.line_no == x.line_no), None)
if al is None:
raise Refused(422, f"the agreement has no line {x.line_no}")
lines_in.append(orders.OrderLineIn(product_id=al.product_id, quantity=x.quantity, unit_code=al.unit_code, unit_price=al.unit_price,
pricing_quantity=al.pricing_quantity))
orders._apply_lines(repo, caller, o, lines_in, today)
for ol, x in zip(o.lines, lines):
ol.agreement_id, ol.agreement_line_no, ol.price_source = agreement_id, x.line_no, "agreement"
o.approved_total = o.net_total
return orders.order_view(repo, caller, o, repo.orders.save(o, 0))
return unit(None, run)
class ScheduleIn(_In):
line_no: int
delivery_date: IsoDate
quantity: Qty
reason: Optional[Reason] = None
def _current(a: rows.PurchaseAgreement) -> list:
top = max([x.revision for x in a.schedule] or [0])
return [x for x in a.schedule if x.revision == top]
@PURCHASING.use_case(kind="command", route="POST /agreements/{agreement_id}/schedule", permission=AGR_WRITE)
def set_schedule(caller: dict, agreement_id: Ref, lines: list[ScheduleIn], version: Optional[Version] = None) -> dict:
"""Set or change delivery dates and quantities of a scheduling agreement (each call is a new revision of the whole schedule). A line dated
before today cannot change; a line already released JIT that falls inside the firm zone needs a reason (the supplier's agreement)."""
def run():
with repository() as repo, _decide():
a, v = _load(repo, caller, agreement_id)
_same_version("the agreement", v, version)
if a.kind != "scheduling_agreement" or a.status != "active":
raise RuleBroken("only an approved scheduling agreement has a schedule")
today = _today(repo)
old = {(x.line_no, x.delivery_date): x for x in _current(a)}
rev = max([x.revision for x in a.schedule] or [0]) + 1
seen = set()
for x in lines:
day = base._day(x.delivery_date, "delivery_date")
key = (x.line_no, day)
seen.add(key)
before = old.get(key)
if before is not None and Decimal(str(before.quantity)) == Decimal(str(x.quantity)):
a.schedule.append(rows.ScheduleLine(a.agreement_id, x.line_no, rev, day, before.quantity, before.unit_code, before.firm, before.changed_reason, before.released_jit))
continue
if day < today or (before is not None and before.delivery_date < today):
raise Refused(422, f"the schedule line of {day} is before today ({today}) and cannot change")
if before is not None and before.released_jit and (day - today).days <= a.firm_zone_days and not (x.reason or "").strip():
raise Refused(409, f"{day} was released JIT and is inside the firm zone ({a.firm_zone_days} days): change it with the supplier's agreement, giving a reason")
al = next((ln for ln in a.lines if ln.line_no == x.line_no), None)
if al is None:
raise Refused(422, f"the agreement has no line {x.line_no}")
a.schedule.append(rows.ScheduleLine(a.agreement_id, x.line_no, rev, day, x.quantity, al.unit_code, bool(before and before.released_jit),
(x.reason or "").strip(), bool(before and before.released_jit)))
for key, was in old.items():
if key not in seen and was.delivery_date >= today:
continue
if key not in seen:
a.schedule.append(rows.ScheduleLine(a.agreement_id, was.line_no, rev, was.delivery_date, was.quantity, was.unit_code, was.firm, was.changed_reason, was.released_jit))
return agreement_view(repo, a, repo.agreements.save(a, v)) | {"schedule": [view(x) for x in _current(a)]}
return unit(version, run)
@PURCHASING.use_case(kind="command", route="POST /agreements/{agreement_id}/releases", permission=AGR_WRITE)
def send_schedule_release(caller: dict, agreement_id: Ref, kind: Literal["jit", "forecast"], to: text(254, min_length=3), through: Optional[IsoDate] = None,
version: Optional[Version] = None) -> dict:
"""Send the schedule as a numbered release (numbered per agreement, the lines as they stand, frozen). A JIT release opens a call-off order keyed
sa:<agreement_id>:<release_no>, confirmed at once under the agreement's approval, so receipts, bills and the match work as for any order; a
FORECAST release commits nothing."""
def run():
with repository() as repo, _decide():
a, v = _load(repo, caller, agreement_id)
_same_version("the agreement", v, version)
if a.kind != "scheduling_agreement" or a.status != "active":
raise RuleBroken("only an approved scheduling agreement sends releases")
today = _today(repo)
end = base._day(through, "through")
lines = [x for x in _current(a) if x.delivery_date >= today and (end is None or x.delivery_date <= end)]
if not lines:
raise RuleBroken("there are no schedule lines to release")
no_ = 1 + max([r.release_no for r in a.releases_sent] or [0])
oid = f"sa:{a.agreement_id}:{no_}" if kind == "jit" else ""
a.releases_sent.append(rows.ScheduleRelease(a.agreement_id, no_, kind, base._stamp(), to, base.j([view(x) for x in lines]), oid, _who(caller)))
if kind == "jit":
for x in lines:
x.released_jit = True
nv = repo.agreements.save(a, v)
out = agreement_view(repo, a, nv) | {"release_no": no_, "call_off_order_id": oid}
if kind == "jit":
o = orders._new_order(repo, caller, oid, a.bp_id, a.currency, source_kind="schedule", source_id=a.agreement_id, agreement_id=a.agreement_id,
approved_by_email=a.approved_by_email, approved_by_sub=a.approved_by_sub, approved_on=a.approved_on)
per = {}
for x in lines:
per[x.line_no] = per.get(x.line_no, ZERO) + x.quantity
lines_in = []
for ln_no, qty in sorted(per.items()):
al = next(ln for ln in a.lines if ln.line_no == ln_no)
lines_in.append(orders.OrderLineIn(product_id=al.product_id, quantity=qty, unit_code=al.unit_code, unit_price=al.unit_price, pricing_quantity=al.pricing_quantity))
orders._apply_lines(repo, caller, o, lines_in, today)
for ol, ln_no in zip(o.lines, sorted(per)):
ol.agreement_id, ol.agreement_line_no, ol.price_source = a.agreement_id, ln_no, "agreement"
o.approved_total = o.net_total
require(orders.confirm_gates(repo, caller, o, today))
o.confirm(today, base.stamp_number(repo, "order", "order_number_format", today, oid))
agreement_claim(repo, o, today)
repo.orders.save(o, 0)
out["call_off_order_id"] = oid
return out
return unit(version, run)
# ── the hooks the order's confirmation calls ───────────────────────────────────────────────────────
def _fmt(x) -> str:
return f"{Decimal(str(x)):,.2f}"
def agreement_gates(repo, o, today: date) -> list:
"""within_agreement: a release order fits what is left of its contract, on a day it is valid."""
if not o.agreement_id:
return []
got = repo.agreements.get(o.agreement_id)
if got is None:
return [no("within_agreement", f"agreement {o.agreement_id} does not exist", "pick another agreement")]
a = got[0]
if a.status != "active":
return [no("within_agreement", f"agreement {a.agreement_no or a.agreement_id} is {a.status}", "approve the agreement")]
if not (a.valid_from <= today <= a.valid_to):
return [no("within_agreement", f"the agreement runs {a.valid_from} to {a.valid_to}; today is {today}", "order on a day it is valid, or revise the agreement")]
c = claimed(repo, a.agreement_id)
if a.kind == "value_contract":
left = fr(a.target_value) - sum((v for _, v in c.values()), Fraction(0))
need = sum((fr(x.net_amount) for x in o.lines), Fraction(0))
if need > left:
return [no("within_agreement", f"{_fmt(as_dec(left, 2))} is left: release up to {_fmt(as_dec(left, 2))}, or revise the contract",
f"release up to {_fmt(as_dec(left, 2))}, or revise the contract")]
else:
for al in a.lines:
need = sum((fr(x.quantity_base) for x in o.lines if x.agreement_line_no == al.line_no), Fraction(0))
left = fr(al.target_quantity) * Fraction(al.unit_numerator, al.unit_denominator) - c.get(al.line_no, (Fraction(0), 0))[0]
if need > left:
return [no("within_agreement", f"line {al.line_no}: {as_dec(max(left, 0), 6).normalize():f} left, {as_dec(need, 6).normalize():f} released",
"release up to what is left, or revise the contract")]
return [ok("within_agreement")]
def agreement_claim(repo, o, today: date) -> None:
"""Claim the release in the same transaction as the confirmation: a claim sequence on the agreement makes two racing confirmations collide, and
the loser re-reads and meets the gate."""
if not o.agreement_id:
return
repo.claims.take(f"agreement:{o.agreement_id}", o.purchase_order_id)
for x in o.lines:
if x.agreement_line_no:
repo.agreement_claims.insert({"agreement_id": o.agreement_id, "line_no": x.agreement_line_no, "purchase_order_id": o.purchase_order_id,
"order_line_no": x.line_no, "quantity_base": x.quantity_base, "value": x.net_amount, "claimed_at": base._stamp()})
orders.AGREEMENT_GATES = agreement_gates
orders.AGREEMENT_CLAIM = agreement_claim