dodil/erp-brick-purchasing
DODIL ERP brick: purchasing
src/brick_purchasing/application/sourcing.py
"""Sourcing: the source list (PUR-44), the quota arrangement (PUR-45) and requests for quotation (PUR-39, PUR-40).
The source list and the quota are dated rows (a change starts AFTER today; the first row of a key may start today). The quota's allocated quantity is
never stored: rules.quota_proposal derives it from confirmed order lines. An RFQ is an aggregate (lines, invitations, quotes); awarding it drafts the
order keyed rfq:<rfq_id>:<bp_id>, with the quoted prices fixed on its lines and the request's approval carried."""
import json
from datetime import date
from decimal import Decimal
from fractions import Fraction
from typing import 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, rules
from brick_purchasing.application.base import (
PURCHASING,
RFQ_WRITE,
SRC_PROPOSE,
SRC_WRITE,
_decide,
_same_version,
_stamp,
_sub,
_today,
_who,
unit,
view,
)
from brick_purchasing.application.inputs import (
Code,
LineNo,
OrderLineIn,
Pct,
Price,
Qty,
Reason,
Ref,
Version,
Whole,
_In,
)
from brick_purchasing.application.master import check_start
from brick_purchasing.domain import rows
from brick_purchasing.domain.errors import Invalid, RuleBroken
from brick_purchasing.domain.values import ZERO, fr, money_round
repository = lambda: base.repository() # noqa: E731
# ═══ the source list (PUR-44) ═══════════════════════════════════════════════════════════════════
@PURCHASING.use_case(kind="command", route="POST /source-list", permission=SRC_WRITE)
def set_source(caller: dict, product_id: Code, bp_id: BigId, valid_from: Optional[IsoDate] = None, fixed: Optional[bool] = None,
blocked: Optional[bool] = None, required: Optional[bool] = None, valid_to: Optional[IsoDate] = None) -> dict:
"""Say which suppliers a product may be bought from, from a date, and block one. `required` makes the list binding for the product (a product
without it may be bought from anyone and the list only proposes); `fixed` marks the preferred source; `blocked` stops orders from that supplier
from the date. Rows are dated and start after today (the first row of a product may start today); a correction is a new row. Ordering from a
blocked or unlisted source is refused with the gate source_allowed."""
def run():
with repository() as repo, _decide():
today = _today(repo)
start = base._day(valid_from, "valid_from") or date.fromordinal(today.toordinal() + 1)
check_start(repo.sources.all(product_id=product_id), start, today, f"source list of {product_id}")
end = base._day(valid_to, "valid_to")
if end and end < start:
raise Invalid("valid_to is before valid_from")
row = {"product_id": product_id, "bp_id": int(bp_id), "valid_from": start, "fixed": bool(fixed), "blocked": bool(blocked), "valid_to": end,
"required": bool(required), "set_by_email": _who(caller)}
repo.sources.put(row)
return row
return unit(None, run)
@PURCHASING.use_case(kind="query", route="GET /source-list/{product_id}", permission=SRC_WRITE)
def get_source_list(caller: dict, product_id: Ref) -> dict:
"""The source list of a product as it stands today: required or not, who is allowed, who is the fixed source, who is blocked from when."""
with repository() as repo:
today = _today(repo)
return {"today": today, "product_id": product_id, **rules.source_info(repo, product_id, today), "rows": repo.sources.all(product_id=product_id)}
# ═══ the quota (PUR-45) ═════════════════════════════════════════════════════════════════════════
class ShareIn(_In):
bp_id: BigId
quota_pct: Pct
base_quantity: Optional[Qty] = None
max_quantity: Optional[Qty] = None
@PURCHASING.use_case(kind="command", route="POST /quotas", permission=SRC_WRITE)
def set_quota(caller: dict, product_id: Code, shares: list[ShareIn], valid_from: Optional[IsoDate] = None) -> dict:
"""Set an agreed split between suppliers for a product from a date (70 / 30). The shares must sum to 100 or the whole arrangement is refused at
entry. The quota rating (allocated + base) / quota is derived from confirmed order lines since the arrangement's valid_from, never a counter."""
def run():
with repository() as repo, _decide():
today = _today(repo)
if not shares:
raise Invalid("give at least one share")
if sum(Decimal(str(s.quota_pct)) for s in shares) != 100:
raise Refused(422, f"the shares sum to {sum(Decimal(str(s.quota_pct)) for s in shares)}, not 100")
if len({s.bp_id for s in shares}) != len(shares):
raise Invalid("a supplier appears twice in the shares")
start = base._day(valid_from, "valid_from") or date.fromordinal(today.toordinal() + 1)
check_start(repo.quotas.all(product_id=product_id), start, today, f"quota of {product_id}")
out = []
for s in shares:
row = {"product_id": product_id, "valid_from": start, "bp_id": int(s.bp_id), "quota_pct": s.quota_pct, "base_quantity": s.base_quantity or ZERO,
"max_quantity": s.max_quantity, "set_by_email": _who(caller)}
repo.quotas.put(row)
out.append(row)
return {"product_id": product_id, "valid_from": start, "shares": out}
return unit(None, run)
@PURCHASING.use_case(kind="query", route="GET /quota-proposal", permission=None)
def propose_source(caller: dict, product_id: Code, quantity: Qty) -> dict:
"""The supplier the quota proposes for a quantity: the lowest rating (allocated + base) / quota, both ratings shown; a blocked source is never
proposed. The buyer may choose another with a reason (kept on the order line). Without a quota in force: no proposal."""
base._require_any(caller, SRC_WRITE, SRC_PROPOSE)
with repository() as repo:
today = _today(repo)
got = rules.quota_proposal(repo, product_id, quantity, today)
return {"today": today, "product_id": product_id, **(got or {"proposed": None, "candidates": []})}
# ═══ requests for quotation (PUR-39, PUR-40) ════════════════════════════════════════════════════
class RfqLineIn(_In):
product_id: Optional[Code] = None
description: Optional[text(500)] = None
quantity: Qty
unit_code: Code
class InviteIn(_In):
bp_id: BigId
to: text(254, min_length=3)
class QuoteLineIn(_In):
line_no: LineNo
unit_price: Price
unit_code: Optional[Code] = None
pricing_quantity: Optional[Whole] = None
delivery_days: Optional[Whole] = None
def _rfq(repo, caller: dict, rfq_id: str) -> tuple:
got = repo.rfqs.get(rfq_id)
if got is None or not auth.has_permission(caller, RFQ_WRITE):
raise Refused(404, f"no RFQ {rfq_id}")
return got
def _quote_net(q: rows.SupplierQuote, r: rows.Rfq) -> Fraction:
total = Fraction(0)
for ql in json.loads(q.lines_json):
x = next(ln for ln in r.lines if ln.line_no == ql["line_no"])
total += fr(ql["unit_price"]) * fr(x.quantity) / int(ql.get("pricing_quantity") or 1)
return total
def _latest_quotes(r: rows.Rfq) -> dict:
best = {}
for q in r.quotes:
if q.bp_id not in best or q.revision > best[q.bp_id].revision:
best[q.bp_id] = q
return best
def rfq_view(r: rows.Rfq, version: int) -> dict:
out = view(r)
out["version"] = version
out["comparison"] = comparison(r)
return out
def comparison(r: rows.Rfq) -> dict:
"""Each line per supplier in the line's unit with the net and delivery days; the cheapest (by net, over suppliers that quoted EVERY line) marked."""
latest = _latest_quotes(r)
complete = {bp: q for bp, q in latest.items() if {ln.line_no for ln in r.lines} <= {x["line_no"] for x in json.loads(q.lines_json)}}
nets = {bp: _quote_net(q, r) for bp, q in complete.items()}
cheapest = min(nets, key=lambda b: (nets[b], b)) if nets else None
suppliers = []
for bp, q in sorted(latest.items()):
suppliers.append({"bp_id": bp, "revision": q.revision, "received_on": q.received_on, "late": q.late, "currency": q.currency,
"complete": bp in complete, "net": str(money_round(nets[bp], r.currency)) if bp in nets else None, "cheapest": bp == cheapest,
"lines": json.loads(q.lines_json)})
return {"suppliers": suppliers, "cheapest_bp_id": cheapest,
"no_quote": sorted(str(i.bp_id) for i in r.invitations if i.bp_id not in latest)}
@PURCHASING.use_case(kind="command", route="POST /rfqs", permission=RFQ_WRITE, idempotent=True)
def create_rfq(caller: dict, deadline: IsoDate, request_id: Optional[Ref] = None, lines: Optional[list[RfqLineIn]] = None,
currency: Optional[text(3, min_length=3)] = None) -> dict:
"""Raise an RFQ, from an APPROVED request (keyed rfq:<request>) or from lines of your own (keyed rfq:<buyer>:<moment>). The deadline is after
today. The RFQ number comes from the series when it is first sent."""
def run():
with repository() as repo, _decide():
today = _today(repo)
dl = base._day(deadline, "deadline")
if dl <= today:
raise Refused(422, f"the deadline {dl} is today or earlier: give a later date")
rid = f"rfq:{request_id}" if request_id else f"rfq:{_who(caller)}:{orders._moment()}"
have = repo.rfqs.get(rid)
if have is not None:
return rfq_view(*have)
cur, rlines, req = currency or "GBP", [], None
if request_id:
got = repo.requests.get(request_id)
if got is None:
raise Refused(404, f"no purchase request {request_id}")
req = got[0]
if req.status != "approved":
raise RuleBroken(f"a request that is {req.status} cannot be sent out for quotes: use an APPROVED request")
cur = currency or req.currency
rlines = [rows.RfqLine(rid, n, x.product_id or "", x.description or "", x.quantity, x.unit_code or "") for n, x in enumerate(req.lines, 1)]
else:
if not lines:
raise Invalid("give the lines to quote, or a request")
rlines = [rows.RfqLine(rid, n, x.product_id or "", x.description or "", x.quantity, x.unit_code) for n, x in enumerate(lines, 1)]
r = rows.Rfq(rid, request_id=request_id or "", deadline=dl, currency=base._currency(cur), created_by_email=_who(caller), created_by_sub=_sub(caller),
lines=rlines)
return rfq_view(r, repo.rfqs.save(r, 0))
return unit(None, run)
@PURCHASING.use_case(kind="command", route="POST /rfqs/{rfq_id}/send", permission=RFQ_WRITE)
def send_rfq(caller: dict, rfq_id: Ref, invitations: Optional[list[InviteIn]] = None, version: Optional[Version] = None) -> dict:
"""Send the RFQ to suppliers: one invitation each, recorded with when and to whom. Sending again records a second send and changes nothing
else. A supplier who is blocked, or not on the source list of a product that needs one, is refused naming the product; the deadline must be
after today. The RFQ number is stamped once, at the first send."""
def run():
with repository() as repo, _decide():
r, v = _rfq(repo, caller, rfq_id)
_same_version("the RFQ", v, version)
if not invitations:
raise Refused(422, "name at least one supplier to invite")
today = _today(repo)
if r.status not in ("draft", "sent"):
raise RuleBroken(f"an RFQ that is {r.status} cannot be sent")
if r.deadline <= today:
raise Refused(422, f"the deadline {r.deadline} is today or earlier: it can no longer be sent")
for inv in invitations:
for x in r.lines:
g = rules.source_gate(repo, x.product_id, inv.bp_id, today)
if not g.met:
raise Refused(409, f"{g.gate}: supplier {inv.bp_id} cannot be invited to quote for {x.product_id}: {g.reason} (fix: {g.fix})")
if not r.rfq_no:
r.rfq_no = base.stamp_number(repo, "rfq", "rfq_number_format", today, r.rfq_id)
now = _stamp()
for inv in invitations:
have = next((i for i in r.invitations if i.bp_id == int(inv.bp_id)), None)
if have is None:
r.invitations.append(rows.RfqInvitation(r.rfq_id, int(inv.bp_id), now, 1, inv.to))
else:
have.sends, have.sent_to = have.sends + 1, inv.to
r.status, r.sent_on = "sent", r.sent_on or today
return rfq_view(r, repo.rfqs.save(r, v))
return unit(version, run)
@PURCHASING.use_case(kind="command", route="POST /rfqs/{rfq_id}/quotes", permission=RFQ_WRITE)
def record_quote(caller: dict, rfq_id: Ref, bp_id: BigId, lines: list[QuoteLineIn], valid_to: Optional[IsoDate] = None,
received_on: Optional[IsoDate] = None, version: Optional[Version] = None) -> dict:
"""Record a supplier's quote (a new revision each time). A quote recorded after the deadline is flagged late (it is still kept and shown)."""
def run():
with repository() as repo, _decide():
r, v = _rfq(repo, caller, rfq_id)
_same_version("the RFQ", v, version)
today = _today(repo)
if r.status != "sent":
raise RuleBroken(f"an RFQ that is {r.status} takes no quotes: send it first" if r.status == "draft" else f"an RFQ that is {r.status} takes no more quotes")
if int(bp_id) not in {i.bp_id for i in r.invitations}:
raise Refused(409, f"supplier {bp_id} was not invited to this RFQ")
got = base._day(received_on, "received_on") or today
if got > today:
raise Invalid("a quote cannot be received after today")
numbers = {x.line_no for x in r.lines}
if any(q.line_no not in numbers for q in lines):
raise Invalid("a quote line names a line the RFQ does not have")
rev = 1 + max([q.revision for q in r.quotes if q.bp_id == int(bp_id)] or [0])
payload = [{"line_no": q.line_no, "unit_price": str(q.unit_price), "unit_code": q.unit_code or next(ln.unit_code for ln in r.lines if ln.line_no == q.line_no),
"pricing_quantity": q.pricing_quantity or 1, "delivery_days": q.delivery_days} for q in lines]
quote = rows.SupplierQuote(r.rfq_id, int(bp_id), rev, got, got > r.deadline, base._day(valid_to, "valid_to"), r.currency, json.dumps(payload), ZERO)
r.quotes.append(quote)
quote.net_total = money_round(_quote_net(quote, r), r.currency) if {x["line_no"] for x in payload} >= numbers else ZERO
return rfq_view(r, repo.rfqs.save(r, v))
return unit(version, run)
@PURCHASING.use_case(kind="query", route="GET /rfqs/{rfq_id}", permission=RFQ_WRITE)
def get_rfq(caller: dict, rfq_id: Ref) -> dict:
"""The RFQ with its invitations and the quotes side by side: each line per supplier in the line's unit, the net, the delivery days, the cheapest
marked, late quotes flagged and the suppliers who have not quoted."""
with repository() as repo:
return rfq_view(*_rfq(repo, caller, rfq_id))
@PURCHASING.use_case(kind="command", route="POST /rfqs/{rfq_id}/award", permission=RFQ_WRITE, idempotent=True)
def award_rfq(caller: dict, rfq_id: Ref, bp_id: BigId, reason: Optional[Reason] = None, version: Optional[Version] = None) -> dict:
"""Award the RFQ to one supplier with a reason (a reason is REQUIRED when the quote is not the cheapest). It drafts the order keyed
rfq:<rfq_id>:<bp_id>, with the quoted prices fixed on its lines and the request's approval carried; awarding again returns the same order. The
unsuccessful suppliers are listed to be told."""
def run():
with repository() as repo, _decide():
r, v = _rfq(repo, caller, rfq_id)
oid = f"rfq:{rfq_id}:{int(bp_id)}"
have = repo.orders.get(oid)
if have is not None:
return {"order": orders.order_view(repo, caller, have[0], have[1]), "rfq": rfq_view(r, v), "to_tell": _to_tell(r)}
_same_version("the RFQ", v, version)
if r.status != "sent":
raise RuleBroken(f"an RFQ that is {r.status} cannot be awarded")
quote = _latest_quotes(r).get(int(bp_id))
if quote is None:
raise Refused(409, f"supplier {bp_id} has not quoted")
cmp_ = comparison(r)
if not next((s for s in cmp_["suppliers"] if s["bp_id"] == int(bp_id)), {}).get("complete"):
raise Refused(409, f"supplier {bp_id}'s quote does not cover every line")
if cmp_["cheapest_bp_id"] != int(bp_id) and not (reason or "").strip():
raise Refused(409, f"supplier {bp_id} is not the cheapest (supplier {cmp_['cheapest_bp_id']} is): give the reason this one is chosen")
today = _today(repo)
req = repo.requests.get(r.request_id)[0] if r.request_id and repo.requests.get(r.request_id) else None
o = orders._new_order(repo, caller, oid, int(bp_id), r.currency, source_kind="rfq", source_id=rfq_id,
**({"request_id": req.purchase_request_id, "approved_total": req.amount, "approved_by_email": req.approved_by_email,
"approved_by_sub": req.approved_by_sub, "approved_on": req.approved_on} if req else {}))
ql = {x["line_no"]: x for x in json.loads(quote.lines_json)}
lines_in = [OrderLineIn(product_id=x.product_id or None, description=x.description or None, quantity=x.quantity, unit_code=ql[x.line_no]["unit_code"] or None,
unit_price=Decimal(ql[x.line_no]["unit_price"]), pricing_quantity=int(ql[x.line_no].get("pricing_quantity") or 1))
for x in r.lines]
orders._apply_lines(repo, caller, o, lines_in, today)
for x in o.lines:
x.price_source = "quote"
r.status, r.awarded_bp_id, r.award_reason, r.order_id = "awarded", int(bp_id), (reason or "").strip(), oid
nv = repo.rfqs.save(r, v)
return {"order": orders.order_view(repo, caller, o, repo.orders.save(o, 0)), "rfq": rfq_view(r, nv), "to_tell": _to_tell(r)}
return unit(version, run)
def _to_tell(r: rows.Rfq) -> list:
return sorted(str(i.bp_id) for i in r.invitations if r.awarded_bp_id is not None and i.bp_id != r.awarded_bp_id)