dodil/erp-brick-purchasing
DODIL ERP brick: purchasing
src/brick_purchasing/application/receipts.py
"""Receipts: what actually arrived, or went back, and when (PUR-07, PUR-25, PUR-26, PUR-53, PUR-58, PUR-62).
POST /purchase-orders/{purchase_order_id}/receipts is the ONE way a receipt is recorded, whether a receiver uses purchasing's screen or
inventory's scanner calls it. Purchasing then hands each line to inventory (POST /receipts; a return: POST /returns-to-supplier) keyed
move_ref pur:<receipt_id>:<line_no> before it answers; it never writes stock. The receipt is a FACT, dated by the delivery (today or earlier)."""
from decimal import Decimal
from fractions import Fraction
from typing import Optional
from erp_platform.bricks import Refused
from erp_platform.types import IsoDate, Limit, text
from brick_purchasing.application import base
from brick_purchasing.application.base import (
PURCHASING,
RCP_READ,
RCP_WRITE,
RECEIPT_HAND,
_decide,
_stamp,
_sub,
_today,
_who,
unit,
view,
)
from brick_purchasing.application.inputs import Offset, OptCode, Reason, ReceiptLineIn, Ref, ReturnLineIn
from brick_purchasing.domain import receipt as rc
from brick_purchasing.domain.errors import Invalid, RuleBroken
from brick_purchasing.domain.order import OrderLine, PurchaseOrder
from brick_purchasing.domain.readiness import no, ok, require, views
from brick_purchasing.domain.values import ZERO, as_dec, base_of, fr, money_round
from brick_purchasing.infrastructure.seams import Missing, ports
repository = lambda: base.repository() # noqa: E731
def _unit_factor(caller: dict, o: PurchaseOrder, x: OrderLine, unit_code: str) -> tuple:
"""(numerator, denominator) of the counted unit to the base unit: the line's own, its pricing unit's, the base unit, else product's."""
if unit_code == x.unit_code:
return x.unit_numerator, x.unit_denominator
if unit_code == x.pricing_unit_code and x.pricing_converts:
return x.pricing_numerator, x.pricing_denominator
if unit_code == x.base_unit_code:
return 1, 1
if x.product_id and ports.installed("product"):
try:
c = ports.product.convert(caller, x.product_id, "1", unit_code, x.base_unit_code)
except Missing as m:
raise Refused(422, f"line {x.line_no}: {unit_code!r} cannot be converted to {x.base_unit_code} exactly: {m.reason}") from None
n, _, d = str(c["exact"]).partition("/")
return int(n), int(d or 1)
raise Refused(422, f"line {x.line_no} is counted in {x.unit_code}: {unit_code!r} cannot be converted")
def qstr(d) -> str:
"""A quantity on the wire: exact, without trailing zeros ('7', not '7.000000')."""
return format(Decimal(str(d)).normalize(), 'f')
def planned_costs(o: PurchaseOrder, x: OrderLine, share: Fraction) -> list:
"""The planned delivery costs of an order line for `share` of what was ordered: [(kind, amount)] — a line's own, plus the order's
(line_no 0) spread by the line's value; a percent is of the line's value."""
total = sum((ln.net_amount for ln in o.lines), ZERO) or Decimal(1)
out = {}
for c in o.delivery_costs:
if c.line_no not in (0, x.line_no):
continue
value = fr(c.amount) if c.amount is not None else (fr(x.net_amount) * fr(c.percent or 0) / 100)
if c.line_no == 0 and c.amount is not None:
value = value * fr(x.net_amount) / fr(total)
out[c.kind] = out.get(c.kind, Fraction(0)) + value * share
return [(k, money_round(v, o.currency)) for k, v in out.items()]
def _line_request(o: PurchaseOrder, r: rc.GoodsReceipt, rl: rc.ReceiptLine, x: OrderLine) -> Optional[dict]:
"""The body for inventory's POST /receipts for this receipt line — or None when no stock is kept for it (a service, an asset, a cost centre
or a project: non-valuated)."""
if x.kind == "service" or x.account_assignment != "stock":
return None
body = {"move_ref": rl.move_ref, "source_kind": {"consignment": "consignment_receipt", "subcontract": "subcontract_receipt"}.get(x.item_category, "purchase_receipt"),
"source_id": r.receipt_id, "purchase_order_id": o.purchase_order_id, "bp_id": str(o.bp_id), "product_id": rl.product_id,
"quantity": qstr(rl.quantity), "unit_code": rl.unit_code, "location_id": rl.location_id, "received_on": r.received_on.isoformat(),
"pricing_unit_code": x.pricing_unit_code, "pricing_numerator": x.pricing_numerator, "pricing_denominator": x.pricing_denominator,
"pricing_quantity": x.pricing_quantity, "currency": o.currency}
share = fr(rl.quantity_base) / fr(x.quantity_base) if x.quantity_base else Fraction(0)
costs = planned_costs(o, x, share)
per = (fr(x.quantity_priced) / x.pricing_quantity) if x.quantity_priced else Fraction(1)
if x.item_category == "consignment":
body["stock_owner_bp_id"] = str(o.bp_id)
return body
landed = fr(x.unit_price) + (sum((fr(a) for _, a in costs), Fraction(0)) / (per * share) if costs and share else Fraction(0))
body["unit_cost"] = str(as_dec(landed, 4))
if costs:
body["delivery_costs"] = [{"kind": k, "amount": str(a)} for k, a in costs]
if x.item_category == "subcontract":
qb = fr(rl.quantity_base)
body["consume_provided"] = [{"product_id": c.product_id, "quantity": qstr(as_dec(fr(c.quantity_per) * qb, 6)), "unit_code": c.unit_code}
for c in o.components if c.line_no == x.line_no]
body["service_fee"] = str(x.unit_price)
return body
def hand_over(caller: dict, receipt_id: str) -> dict:
"""Tell inventory what arrived (or went back), per line not yet handed over. Inventory's refusal leaves the receipt standing and the line
showing its reason; a retry (the clock worker, or hand_receipt_to_inventory) picks it up. Idempotent on move_ref."""
with repository() as repo:
got = repo.receipts.get(receipt_id)
og = repo.orders.get(got[0].purchase_order_id) if got else None
if not got or not og:
raise Refused(404, f"no receipt {receipt_id}")
r, o = got[0], og[0]
results = {}
for rl in r.lines:
if rl.handed_over_at is not None:
continue
x = o.line(rl.order_line_no)
if r.kind == "return":
body = _return_request(o, r, rl, x)
else:
body = _line_request(o, r, rl, x) if x else None
if body is None:
results[rl.line_no] = ("kept", "", "no stock is kept for this line (a service, an asset, a cost centre or a project)")
elif not ports.installed("inventory"):
results[rl.line_no] = ("kept", "", "inventory is not installed: the receipt stands without stock")
else:
try:
got_ = ports.inventory.return_to_supplier(body) if r.kind == "return" else ports.inventory.receive(body)
results[rl.line_no] = ("moved", str(got_.get("move_id") or ""), "")
except Missing as m:
gates = "; ".join(f"{g.get('gate')}: {g.get('reason')}" for g in m.gates) or m.reason
results[rl.line_no] = ("refused", "", f"inventory refused ({m.status}): {gates}")
def apply():
with repository() as repo2, _decide():
r2, v2 = repo2.receipts.get(receipt_id)
for rl in r2.lines:
res = results.get(rl.line_no)
if not res:
continue
if res[0] in ("kept", "moved"):
rl.handed_over_at, rl.stock_move_id, rl.handover_note = _stamp(), res[1], res[2]
else:
rl.handover_note = res[2]
r2.settle_status()
repo2.receipts.save(r2, v2)
return r2
return unit(None, apply)
def _return_request(o: PurchaseOrder, r: rc.GoodsReceipt, rl: rc.ReceiptLine, x: Optional[OrderLine]) -> Optional[dict]:
if x is None or x.kind == "service" or x.account_assignment != "stock":
return None
orig = r.corrects_receipt_id
return {"move_ref": rl.move_ref, "reverses_move_ref": f"pur:{orig}:{rl.reverses_line_no}", "source_kind": "purchase_return", "source_id": r.receipt_id,
"bp_id": str(o.bp_id), "product_id": rl.product_id, "quantity": qstr(abs(rl.quantity)), "unit_code": rl.unit_code, "location_id": rl.location_id,
"returned_on": r.received_on.isoformat()}
def receipt_view(r: rc.GoodsReceipt, version: Optional[int] = None, gates: Optional[list] = None) -> dict:
out = view(r)
out["lines"] = [{**ln, "handed_over": ln["handed_over_at"] is not None} for ln in out["lines"]]
if version is not None:
out["version"] = version
if gates is not None:
out["readiness"] = views(gates)
return out
def _answer(r: rc.GoodsReceipt, gates: list, replay: bool = False) -> dict:
return {"receipt_id": r.receipt_id, "kind": r.kind, "status": r.status, "replay": replay,
"lines": [{"line_no": ln.line_no, "order_line_no": ln.order_line_no, "quantity": ln.quantity, "quantity_base": ln.quantity_base,
"move_ref": ln.move_ref, "stock_move_id": ln.stock_move_id, "handover_note": ln.handover_note} for ln in r.lines],
"readiness": views(gates)}
@PURCHASING.use_case(kind="command", route="POST /purchase-orders/{purchase_order_id}/receipts", permission=RCP_WRITE, idempotent=True)
def record_goods_receipt(caller: dict, purchase_order_id: Ref, delivery_note: text(100, min_length=1), received_on: IsoDate, lines: list[ReceiptLineIn],
location_id: Optional[OptCode] = None) -> dict:
"""Record what actually arrived on a confirmed order: the delivery note (the idempotency key — recording DN-881 again receives nothing twice),
the delivery's own date (today or earlier, never later), the location and, per order line, the quantity and unit counted (the pack, or the base
unit). The quantity is also kept in the base unit. Purchasing then hands each line to inventory before it answers; with inventory not installed
the receipt still stands. Catch weight arrives in M7 and is refused until then."""
if not lines:
raise Refused(422, "a receipt needs at least one line")
for ln in lines:
try:
rc.check_catch(ln.model_dump())
except Invalid as e:
raise Refused(422, str(e)) from None
rid = rc.receipt_id(purchase_order_id, delivery_note.strip())
def record():
with repository() as repo, _decide():
have = repo.receipts.get(rid)
if have:
return _answer(have[0], [ok("not_future")], replay=True), True
got = repo.orders.get(purchase_order_id)
if got is None or not base.auth.has_permission(caller, RCP_WRITE):
raise Refused(404, f"no purchase order {purchase_order_id}")
o = got[0]
if o.status != "confirmed":
raise RuleBroken(f"the order is {o.status}: goods are received against a confirmed order")
today = _today(repo)
when = base._day(received_on, "received_on")
gates = [ok("not_future") if when <= today else no("not_future", f"a delivery cannot be dated after today ({today})", "record it on the day it arrives")]
needs_loc = repo.settings.value_on("receipt_needs_location", today) and ports.installed("inventory")
seen = set()
r = rc.GoodsReceipt(rid, "receipt", purchase_order_id, delivery_note.strip(), when, _who(caller), _sub(caller), location_id or "", "", "", "recorded", o.bp_id)
for n, ln in enumerate(lines, 1):
x = o.line(ln.line_no)
if x is None:
raise Refused(422, f"the order has no line {ln.line_no}")
if ln.line_no in seen:
raise Refused(422, f"line {ln.line_no} appears twice in this delivery")
seen.add(ln.line_no)
if ln.quantity <= 0:
raise RuleBroken(f"each quantity is above zero (line {ln.line_no})")
u = ln.unit_code or x.unit_code
nu, de = _unit_factor(caller, o, x, u)
qb = base_of(ln.quantity, nu, de)
loc = location_id or ""
if needs_loc and x.kind == "goods" and x.account_assignment == "stock" and not loc:
gates.append(no("location", "inventory is installed and the line has no location", "pick the receiving location"))
r.lines.append(rc.ReceiptLine(rid, n, x.line_no, x.product_id, ln.quantity, u, as_dec(qb), loc, f"pur:{rid}:{n}"))
if not any(g.gate == "location" for g in gates):
gates.append(ok("location"))
require(gates)
repo.receipts.save(r, 0)
return _answer(r, gates), False
answer, replay = unit(None, record)
if replay:
return answer
r2 = hand_over(caller, rid)
return _answer(r2, [ok("not_future"), ok("location")])
@PURCHASING.use_case(kind="command", route="POST /receipts/{receipt_id}/returns", permission=RCP_WRITE)
def record_return(caller: dict, receipt_id: Ref, lines: list[ReturnLineIn], reason: Reason) -> dict:
"""Send goods back to the supplier, against a receipt, with a reason: a reversing receipt row (kind return) dated today; the original is
untouched. A return larger than what was received (less earlier returns) on that delivery is refused. Inventory is told to move the stock
out (POST /returns-to-supplier); if the stock is no longer at the location, inventory refuses, the return stands and the hand-over shows why."""
if not lines or not reason.strip():
raise Refused(422, "a return needs at least one line and a reason")
def record():
with repository() as repo, _decide():
got = repo.receipts.get(receipt_id)
if got is None or got[0].kind != "receipt":
raise Refused(404, f"no receipt {receipt_id}")
orig = got[0]
og = repo.orders.get(orig.purchase_order_id)
o = og[0]
earlier = [x for x in repo.receipts.of_order(orig.purchase_order_id) if x.kind == "return" and x.corrects_receipt_id == receipt_id]
rid = rc.return_id(receipt_id, len(earlier) + 1)
today = _today(repo)
ret = rc.GoodsReceipt(rid, "return", orig.purchase_order_id, f"return of {orig.delivery_note}", today, _who(caller), _sub(caller), orig.location_id,
reason.strip(), receipt_id, "recorded", o.bp_id)
for n, ln in enumerate(lines, 1):
src = orig.line(ln.line_no)
if src is None:
raise Refused(422, f"the receipt has no line {ln.line_no}")
x = o.line(src.order_line_no)
back = sum((-fr(rl.quantity_base) for e in earlier for rl in e.lines if rl.reverses_line_no == ln.line_no), Fraction(0))
nu, de = _unit_factor(caller, o, x, src.unit_code)
asked = base_of(ln.quantity, nu, de)
rc.check_return(fr(src.quantity_base), back, asked, ln.line_no)
rl = rc.ReceiptLine(rid, n, src.order_line_no, src.product_id, -ln.quantity, src.unit_code, -as_dec(asked), src.location_id, f"pur:{rid}:{n}",
reverses_line_no=ln.line_no)
ret.lines.append(rl)
repo.receipts.save(ret, 0)
return rid
rid = unit(None, record)
r2 = hand_over(caller, rid)
out = _answer(r2, [ok("within_received")])
return out
@PURCHASING.use_case(kind="command", route="POST /receipts/{receipt_id}/hand-over", permission=RECEIPT_HAND, idempotent=True)
def hand_receipt_to_inventory(caller: dict, receipt_id: Ref) -> dict:
"""Retry telling inventory about the lines of a receipt it has not taken yet (the composition's clock worker calls this). Idempotent: inventory
records one move per move_ref."""
r = hand_over(caller, receipt_id)
return _answer(r, [ok("inventory")])
@PURCHASING.use_case(kind="query", route="GET /receipts/{receipt_id}", permission=RCP_READ)
def get_goods_receipt(caller: dict, receipt_id: Ref) -> dict:
"""One receipt (or return) with its lines, the move id inventory answered and, where a hand-over is pending, the reason."""
with repository() as repo:
got = repo.receipts.get(receipt_id)
if got is None:
raise Refused(404, f"no receipt {receipt_id}")
return receipt_view(got[0], got[1])
@PURCHASING.use_case(kind="query", route="GET /receipts", permission=RCP_READ)
def list_goods_receipts(caller: dict, purchase_order_id: Optional[Ref] = None, limit: Limit = 50, offset: Offset = 0) -> dict:
"""Receipts and returns (heads), newest first; for one order when given."""
with repository() as repo:
rows, total = repo.receipts.page(purchase_order_id=purchase_order_id, limit=limit, offset=offset)
return {"total": total, "receipts": [{k: getattr(r, k) for k in ("receipt_id", "kind", "purchase_order_id", "delivery_note", "received_on", "status")}
for r in rows]}
@PURCHASING.use_case(kind="query", route="GET /receiving-locations", permission=RCP_WRITE)
def list_receiving_locations(caller: dict, warehouse_id: Optional[OptCode] = None) -> dict:
"""The receiving screen's location picker, asked of inventory (usage receiving)."""
try:
got = ports.inventory.locations("receiving", warehouse_id or None)
except Missing as m:
raise base._refuse(m, "inventory") from None
got.pop("_status", None)
return got